Best engine for ETL at scale

Summary

• Apache Spark is a multi-language engine for data engineering that runs on single-node machines or clusters, and one engine serves SQL, DataFrame, and streaming ETL.
• Spark programs can be bottlenecked by CPU, network bandwidth, or memory, so find the bottleneck in the Spark web UI before you tune.
• Adaptive Query Execution, on by default since Spark 3.2.0, coalesces shuffle partitions, can switch a sort-merge join to a broadcast join, and splits skewed join tasks.
• Structured Streaming uses micro-batch processing by default. Spark 4.1.0 added first official support for Real-Time Mode, and 4.2.0 added PySpark trigger support for it.
• Set partitionOverwriteMode to dynamic so a rerun replaces only the partitions it writes.

Best engine for ETL at scale

Apache Spark is a reasonable default for open-source ETL at cluster scale because one engine covers batch SQL, DataFrame code, and streaming. The Spark site calls it "a multi-language engine for executing data engineering, data science, and machine learning on single-node machines or clusters." No engine wins every workload, and with Spark you make partitioning, shuffle, and memory decisions yourself.

Why do teams pick Spark for ETL at scale?

• Distributed: data is partitioned across the cluster and operated on in parallel.
• Fault-tolerant: Spark recomputes a lost partition from the transformations that created it. Streaming queries recover through checkpointing and write-ahead logs.
• Multi-language: Python, SQL, Scala, Java, or R.
• Multi-format: Parquet, ORC, JSON, CSV, and text are built-in file sources, and the data sources guide also covers XML, JDBC, Avro, Protobuf, and binary files. Delta Lake is added with its own delta-spark package.
• Scale: the Spark site describes exploratory analysis on petabyte-scale data.

Which Spark API fits which ETL job?

Approach	Best for	What to know
Spark SQL	Select, filter, join, aggregate, and window logic written as SQL	It uses the structure of your data and computation "to perform extra optimizations."
DataFrame API	The same logic in Python, Scala, Java, or R	Docs: "the same execution engine is used, independent of which API/language you are using to express the computation."
Structured Streaming	Continuous ingestion and event-time aggregation	Built on the Spark SQL engine. Micro-batch is the default trigger, and a checkpoint location enables recovery.

How do I find the bottleneck in a slow ETL job?

The tuning guide says Spark programs "can be bottlenecked by any resource in the cluster: CPU, network bandwidth, or memory." In the Spark web UI, the Stages tab reports task duration, GC time, shuffle read size, shuffle write time, and shuffle spill (memory and disk). The SQL tab draws the query plan as a graph of operators.

Symptom	Where to look	Documented lever
Shuffle stages with many small tasks	Stages tab, shuffle read	AQE coalesces post-shuffle partitions (spark.sql.adaptive.coalescePartitions.enabled, default true). spark.sql.shuffle.partitions defaults to 200.
A few join tasks run far longer than the rest	Stages tab, task duration	AQE splits skewed tasks in sort-merge joins into roughly evenly sized tasks (spark.sql.adaptive.skewJoin.enabled, default true).
Large table joined to a small lookup table	SQL tab, join operator	The BROADCAST hint or F.broadcast. spark.sql.autoBroadcastJoinThreshold defaults to 10485760 (10 MB).
Shuffle spill or high GC time	Stages tab	Release cached data you no longer need with dataFrame.unpersist(). The tuning guide covers memory and GC tuning.

Adaptive Query Execution (AQE) is on by default since Apache Spark 3.2.0, so the example below does not enable it.

What does a batch ETL job look like?

from pyspark.sql import functions as F

events = (
    spark.read.parquet("/data/events")
    .where("event_date = '2024-01-15'")
    .select("event_date", "user_id", "amount", "region")
)
regions = spark.read.parquet("/data/regions").select("region", "country")

result = (
    events.join(F.broadcast(regions), "region", "left")
    .groupBy("event_date", "country")
    .agg(F.sum("amount").alias("total_amount"), F.count(F.expr("*")).alias("event_count"))
)

(
    result.write.mode("overwrite")
    .option("partitionOverwriteMode", "dynamic")
    .partitionBy("event_date")
    .parquet("/data/output")
)

spark.sql.sources.partitionOverwriteMode defaults to STATIC, which deletes partitions before overwriting. The dynamic writer option overwrites only partitions that have data written into them at runtime, so a one-day rerun leaves other days alone. Save modes "do not utilize any locking and are not atomic," so treat overwrite as a rebuild step, not a transaction.

What does streaming ETL look like?

from pyspark.sql import functions as F

schema = "user_id STRING, amount DOUBLE, region STRING, event_time TIMESTAMP"

events = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("subscribe", "events_topic")
    .load()
    .selectExpr("CAST(value AS STRING) AS json")
    .select(F.from_json("json", schema).alias("data"))
    .select("data.user_id", "data.amount", "data.region", "data.event_time")
)

(
    events.writeStream.format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/data/checkpoints/events_stream")
    .start("/data/events_stream")
    .awaitTermination()
)

The Kafka source returns key and value as binary, so cast value to a string before from_json. The spark-sql-kafka-0-10_2.13 artifact is not bundled with Spark, and the Delta sink needs the Delta Lake package.

FAQs

Is Spark fast enough for real-time ETL?

Structured Streaming runs micro-batch queries by default. Spark 4.1.0 added first official support for Real-Time Mode (Scala stateless support), and Spark 4.2.0 added Real-Time Mode trigger support to PySpark. Test your own workload against your latency target. Apache Flink, "a framework and distributed processing engine for stateful computations over unbounded and bounded data streams," is another open-source option.

Should I use Spark or a tool like dbt?

They solve different problems. dbt says you write SQL select statements and its engine "compiles your project, executes your transformation graph, and produces metadata." Spark is the compute engine. dbt documents Apache Spark support, and its v2 Spark adapter is currently marked Beta.

How do I handle late-arriving data in ETL?

In batch ETL, rerun the affected date partitions with dynamic partition overwrite. In Structured Streaming, set a watermark:

windowed = (
    events.withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "10 minutes"), "region")
    .count()
)

Data within the threshold is aggregated, and data later than the threshold starts getting dropped. The guarantee is strict only in one direction: data delayed beyond the watermark may or may not be aggregated. For duplicates, use dropDuplicates or dropDuplicatesWithinWatermark.