Apache Spark: RDDs, Transformations & Actions

RDD Basics

RDD (Resilient Distributed Dataset) is the low-level foundation of Spark. Prefer DataFrames for most use cases — they are more optimized. Use RDDs for unstructured data or custom partitioning logic.

sc = spark.sparkContext

# Create RDDs
rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=4)
rdd = sc.textFile("hdfs:///data/logs/*.log")
rdd = sc.wholeTextFiles("hdfs:///data/files/")  # (filename, content) pairs

# Transformations (lazy — create new RDD, not computed yet)
rdd.map(lambda x: x * 2)
rdd.flatMap(lambda line: line.split(" "))    # one element → zero or more
rdd.filter(lambda x: x > 3)
rdd.distinct()
rdd.sample(withReplacement=False, fraction=0.1)
rdd.union(rdd2)
rdd.intersection(rdd2)
rdd.subtract(rdd2)
rdd.sortBy(lambda x: x, ascending=False)
rdd.repartition(8)       # shuffle and repartition
rdd.coalesce(4)          # reduce partitions without shuffle

# Key-value transformations
pairs = rdd.map(lambda x: (x % 3, x))   # (key, value) RDD
pairs.groupByKey()
pairs.reduceByKey(lambda a, b: a + b)    # more efficient than groupByKey
pairs.sortByKey()
pairs.mapValues(lambda v: v * 2)
pairs.join(other_pairs)
pairs.leftOuterJoin(other_pairs)
pairs.cogroup(other_pairs)

# Actions (trigger computation)
rdd.collect()            # bring all data to driver (careful with large datasets)
rdd.count()
rdd.first()
rdd.take(10)
rdd.top(10)
rdd.sum()
rdd.min(), rdd.max()
rdd.mean(), rdd.variance()
rdd.reduce(lambda a, b: a + b)
rdd.countByValue()
rdd.saveAsTextFile("output/")
rdd.foreach(print)

Partitioning & Persistence

from pyspark import StorageLevel

# Check partitions
rdd.getNumPartitions()
df.rdd.getNumPartitions()

# Repartition (use when data is skewed)
df.repartition(200)                    # shuffle, creates even partitions
df.repartition(200, "user_id")         # partition by column (colocates same user_id)
df.coalesce(50)                        # reduce partitions (no shuffle)

# Persist / cache (avoid recomputing expensive transformations)
df.cache()                             # = persist(MEMORY_AND_DISK)
df.persist(StorageLevel.MEMORY_ONLY)
df.persist(StorageLevel.DISK_ONLY)
df.persist(StorageLevel.MEMORY_AND_DISK)

# Unpersist when done
df.unpersist()

# Custom partitioner (RDD only)
pairs.partitionBy(8, lambda key: hash(key) % 8)

# Check data distribution
df.groupBy(F.spark_partition_id()).count().show()  # records per partition

Broadcast Variables & Accumulators

# Broadcast: send large lookup table to all workers once (not per task)
country_map = spark.sparkContext.broadcast({
    "US": "United States", "DE": "Germany", "JP": "Japan"
})

df.withColumn("country_name",
    F.udf(lambda code: country_map.value.get(code, "Unknown"))(F.col("country_code"))
)

# Better: use DataFrame join with broadcast hint instead of UDF
country_df = spark.createDataFrame([("US", "United States"), ("DE", "Germany")],
    ["code", "name"])
df.join(F.broadcast(country_df), df.country_code == country_df.code, "left")

# Accumulator: collect metrics from workers
error_count = spark.sparkContext.accumulator(0)

def process(row):
    global error_count
    try:
        parse(row)
    except Exception:
        error_count += 1

rdd.foreach(process)
print(f"Errors: {error_count.value}")