How do I migrate MapReduce jobs to Spark?

Summary

• Rewrite each MapReduce stage as a Spark transformation, then prefer DataFrames. Mappers become map(), flatMap() and filter(). Reducers become reduceByKey() or aggregateByKey().
• reduceByKey() already does a combiner's job. The PySpark docs say it merges values locally on each mapper before sending results to a reducer, similarly to a combiner in MapReduce.
• Spark still shuffles, and a shuffle still touches disk. The Spark guide says it involves disk I/O, data serialization and network I/O.
• Counters become accumulators, and the distributed cache becomes broadcast variables or addFile().
• On Databricks, serverless compute doesn't support the RDD APIs, so target DataFrames and Spark SQL.

How do I migrate MapReduce jobs to Spark?

Map each MapReduce concept to its Spark equivalent, then prefer DataFrames and Spark SQL for new code. You can keep your data and cluster manager. Spark uses Hadoop's client libraries for HDFS and YARN, and PySpark can read any Hadoop InputFormat or write any Hadoop OutputFormat, so existing formats survive the rewrite.

What are the core differences?

In MapReduce, the Mapper outputs are sorted and partitioned per Reducer, and a Reducer has three primary phases: shuffle, sort and reduce. Spark instead builds up lazy transformations. Transformations don't compute anything until an action needs a result at the driver, which lets Spark, for example, return only the result of a reduce rather than the larger mapped dataset.
Spark has a shuffle too. Operations such as repartition, groupByKey, reduceByKey and join can cause one. The guide calls the shuffle expensive because it involves disk I/O, data serialization and network I/O, and it generates intermediate files on disk. It also notes that the "map" and "reduce" task names in the shuffle come from MapReduce and don't directly relate to Spark's map and reduce operations.

How do MapReduce concepts map to Spark?

MapReduce	Spark	Notes
Mapper map()	map(), flatMap(), filter()	flatMap() can return zero or more outputs per input, like a Mapper
Mapper setup() and cleanup()	mapPartitions()	The function runs once per partition
Reducer	reduceByKey(), aggregateByKey()	The guide recommends them over groupByKey() for aggregations
Combiner	reduceByKey()	Merges locally on each mapper before the shuffle. aggregateByKey() allows a different result type
Partitioner and sort	partitionBy(), sortByKey(), repartitionAndSortWithinPartitions()	The last one can push sorting down into the shuffle machinery
Counters	Accumulators	Updates inside actions are applied once per task. Inside transformations they may be applied more than once
DistributedCache	Broadcast variables, SparkContext.addFile()	Read-only values or files made available on every node
MultipleOutputs	Separate writes, or partitionBy on write

What does a migrated job look like?

A word count shows the shape. The RDD version mirrors the MapReduce structure:

# sc is predefined in the PySpark shell
counts = (sc.textFile("input.txt")
            .flatMap(lambda line: line.split())
            .map(lambda word: (word, 1))
            .reduceByKey(lambda a, b: a + b))
counts.saveAsTextFile("output")

For new code, prefer the DataFrame API. The Spark SQL guide says its interfaces give Spark more information about the structure of the data and the computation than the basic RDD API does, and Spark uses that to perform extra optimizations. spark.read.text() returns one string column named value. The text writer accepts only a single string column, so write the two-column result as Parquet.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("WordCount").getOrCreate()
words = spark.read.text("input.txt").select(
    F.explode(F.split("value", " ")).alias("word"))
words.groupBy("word").count().write.mode("overwrite").parquet("output")

When the aggregate has a different type than the values, such as an average, use aggregateByKey() with a sequence function and a combine function:

pairs = sc.parallelize([("a", 1), ("a", 3), ("b", 5)])
sums = pairs.aggregateByKey((0, 0),
                            lambda acc, v: (acc[0] + v, acc[1] + 1),
                            lambda a, b: (a[0] + b[0], a[1] + b[1]))
averages = sums.mapValues(lambda t: t[0] / t[1])

How does Databricks approach migrating MapReduce jobs?

The Databricks ETL migration guide says pipelines written in languages other than SQL, Apache Spark or Hive might need heavy refactoring, so plan to rewrite MapReduce code as Spark. Databricks creates a SparkSession for each notebook or job, so use SparkSession.builder.getOrCreate() as above instead of creating new sessions, and remove terminal commands such as sc.stop(). Serverless compute supports only Spark Connect APIs and doesn't support the RDD APIs, so use the DataFrame version. Databricks recommends Delta Lake instead of Parquet or ORC when writing data, and Lakeflow Jobs orchestrate multi-task workflows. Photon is the Databricks-native vectorized query engine and is compatible with Apache Spark APIs.

FAQs

Does Spark still write intermediate data to disk?

At shuffles, yes: the shuffle generates intermediate files on disk. Spark also automatically persists some intermediate shuffle data, such as for reduceByKey, to avoid recomputing the whole input if a node fails during the shuffle.

How does Spark recover from a failed node?

RDDs automatically recover from node failures. If a persisted partition is lost, Spark recomputes it using the transformations that originally created it.

Can I run Spark on my existing Hadoop cluster?

Yes. Spark runs on YARN, for example with spark-submit --master yarn --deploy-mode cluster, and HADOOP_CONF_DIR or YARN_CONF_DIR must point to the Hadoop client configuration.