Dataproc vs Dataflow: Which to Choose

Two Google Cloud services transform large amounts of data, and newcomers are regularly asked to choose between them before they understand either. Dataproc runs Apache Spark and the Hadoop family of tools. Dataflow runs pipelines written with Apache Beam. Both read from Cloud Storage and write to BigQuery, both scale to large data, and both can do batch and streaming. The choice is rarely about capability. It is about what you already have and what you want to operate.

One point of naming first. Google has renamed Dataproc: it is now Managed Service for Apache Spark (Google Cloud: Managed Service for Apache Spark). The command-line tool still says dataproc, and most people and job adverts still use the old name. This guide uses "the Spark service" for it.

At a glance

The Spark service has two forms

The rename brought two products under one name (Google Cloud: compare serverless and cluster deployments):

• Clusters. You create a cluster of virtual machines, and it runs Spark along with other open source engines such as Hive, Flink and Trino. You can connect to the machines, tune them and install what you need. You pay while the cluster is up: a management fee plus the machines.
• Serverless. You submit a Spark job and the service finds the capacity. There is no cluster to size or shut down, and you pay for the job's run time, per second (Google Cloud: pricing).

Google's guidance on choosing between them is clear. Clusters suit migrating existing Hadoop and Spark workloads, long-running shared environments and anything needing fine control. Serverless suits new pipelines, scheduled batch jobs and irregular demand.

That distinction changes the old comparison. "Dataproc means managing clusters" is no longer true by necessity.

Dataflow has one form

Dataflow is serverless only. You submit a pipeline; it starts workers, scales them with the load and removes them at the end (Google Cloud: Dataflow overview). Its engine, Apache Beam, was designed around streaming: windows defined on the time events happened, watermarks that track lateness, and exactly-once processing by default (Google Cloud: streaming pipelines). Batch is treated as the special case of a stream that ends.

The same job, twice

To make the difference in programming model concrete, the script answers one question both ways: on 15 January 2024, which three hours had the most New York taxi pickups, and what was the average fare in each (NYC TLC: trip record data)? Both versions run locally, Spark in local mode and Beam on its Direct Runner. Neither is a run on the Google Cloud services, and the comparison is of code, not speed.

# One small job written twice: with PySpark, as it would run on the Spark service (formerly
# Dataproc), and with Apache Beam, as it would run on Dataflow. Both run locally here, Spark in
# local mode and Beam on the Direct Runner. This is not a run on either Google Cloud service.
import glob
import os
import shutil
import sys

import apache_beam as beam
import pandas as pd
from apache_beam.options.pipeline_options import PipelineOptions
from pyspark.sql import SparkSession, functions as F

SOURCE = "../../dedata/yellow_tripdata_2024-01.parquet"
DAY = "2024-01-15"                       # question: the three busiest pickup hours, and their average fare

# --- PySpark: a DataFrame query, planned and optimized by the engine --------------------------------
os.environ["PYSPARK_PYTHON"] = sys.executable
spark = SparkSession.builder.master("local[2]").appName("busiest-hours").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
spark_rows = (spark.read.parquet(SOURCE)
              .where((F.to_date("tpep_pickup_datetime") == DAY) & (F.col("fare_amount") > 0))
              .groupBy(F.hour("tpep_pickup_datetime").alias("hour"))
              .agg(F.count("*").alias("trips"), F.round(F.avg("fare_amount"), 2).alias("avg_fare"))
              .orderBy(F.desc("trips")).limit(3).collect())
spark_result = [(row.hour, row.trips, row.avg_fare) for row in spark_rows]
spark.stop()

# --- Apache Beam: a pipeline of transforms over elements ---------------------------------------------
trips = pd.read_parquet(SOURCE, columns=["tpep_pickup_datetime", "fare_amount"])
day = trips[trips.tpep_pickup_datetime.dt.strftime("%Y-%m-%d") == DAY]
records = [(row.tpep_pickup_datetime.hour, float(row.fare_amount)) for row in day.itertuples()]

shutil.rmtree("out", ignore_errors=True)
with beam.Pipeline(options=PipelineOptions(["--runner=DirectRunner"])) as pipeline:
    (pipeline
     | "Read" >> beam.Create(records)
     | "PositiveFare" >> beam.Filter(lambda pair: pair[1] > 0)
     | "PairUp" >> beam.Map(lambda pair: (pair[0], (pair[1], pair[1])))
     | "CountAndMean" >> beam.CombinePerKey(beam.combiners.TupleCombineFn(
           beam.combiners.CountCombineFn(), beam.combiners.MeanCombineFn()))
     | "Format" >> beam.Map(lambda pair: f"{pair[1][0]:06d} {pair[0]} {pair[1][1]:.2f}")
     | "Write" >> beam.io.WriteToText("out/hours", num_shards=1, shard_name_template=""))
lines = sorted((line.split() for path in glob.glob("out/hours*") for line in open(path)), reverse=True)[:3]
beam_result = [(int(hour), int(count), float(mean)) for count, hour, mean in lines]

print(f"{'engine':8} {'hour':>4} {'trips':>6} {'average fare':>13}")
for engine, result in (("PySpark", spark_result), ("Beam", beam_result)):
    for hour, count, mean in result:
        print(f"{engine:8} {hour:4} {count:6,} {mean:13.2f}")
print(f"same answer from both: {spark_result == beam_result}")

engine   hour  trips  average fare
PySpark    14  5,877         19.95
PySpark    15  5,724         19.10
PySpark    17  5,669         19.18
Beam       14  5,877         19.95
Beam       15  5,724         19.10
Beam       17  5,669         19.18
same answer from both: True

Both engines agree: the busiest hours were 14:00, 15:00 and 17:00. What differs is how you say it.

• The Spark version is a query. Filter, group, aggregate, order, limit. It reads like SQL because it nearly is, and Spark's optimizer decides how to execute it. Anyone who knows SQL or pandas can follow it.
• The Beam version is a pipeline. Each line is a transform applied to elements: filter, pair up, combine per key, format, write. It is more explicit and a little longer. In exchange, the same shape extends naturally to unbounded data, where you would add a window before the combine and nothing else would change.

That is the trade in miniature. Spark is the more comfortable tool for table-shaped batch analysis. Beam asks more of you up front and repays it when the data never stops arriving.

Submitting each

# Spark job on a cluster
gcloud dataproc jobs submit pyspark gs://example-code/busiest_hours.py \
  --cluster=analytics-cluster \
  --region=asia-south1

# Beam pipeline on Dataflow
python busiest_hours_beam.py \
  --runner DataflowRunner \
  --project analytics-prod \
  --region asia-south1 \
  --temp_location gs://example-dataflow/tmp/

The commands follow the documented forms (Google Cloud: submit a job; Google Cloud: create a pipeline using Python).

How to choose

1. Do you have Spark or Hadoop code already? Use the Spark service. Rewriting working jobs in Beam to change services is seldom worth it. Use clusters for a faithful migration and serverless for new jobs.
2. Is the team fluent in Spark? The same answer. Skills are the largest cost in any pipeline.
3. Is the workload streaming, with event time, late data or exactly-once requirements? Use Dataflow. This is what Beam was built for.
4. Do you want nothing to operate, and are you starting fresh? Dataflow, or the serverless Spark deployment. Either avoids clusters.
5. Do you need other engines beside Spark, such as Hive, Trino or Flink? Only clusters offer them.
6. Is the transformation expressible in SQL on data already in BigQuery? Then perhaps neither. A scheduled query or a SQL transformation tool is simpler than either engine. See scheduling queries in BigQuery.

Many platforms use both: Dataflow for the streaming path into the warehouse, Spark for heavy batch jobs and data science work.

Mistakes to avoid

• Leaving a cluster running. Cluster billing is for uptime, whether or not jobs run. Delete it, schedule its deletion, or use serverless.
• Choosing Dataflow for a job that is one SQL statement.
• Rewriting for fashion. A stable Spark pipeline does not become better by being ported.
• Judging by the old name. Tutorials about Dataproc describe the cluster deployment. The serverless option may not appear in them at all.

To learn each in more depth, see the Dataflow tutorial and the PySpark tutorial. For where both fit, see the GCP roadmap.

Spark, Beam and pipeline design are covered in the Data Engineering course.

Sources

• Managed Service for Apache Spark on clusters overview (Google Cloud Documentation) — Google Cloud
• Compare Managed Service for Apache Spark serverless and cluster deployments (Google Cloud Documentation) — Google Cloud
• Managed Service for Apache Spark (formerly Dataproc) pricing (Google Cloud) — Google Cloud
• Submit a job (Google Cloud Documentation) — Google Cloud
• Dataflow overview (Google Cloud Documentation) — Google Cloud
• Streaming pipelines (Google Cloud Documentation) — Google Cloud
• Create a Dataflow pipeline using Python (Google Cloud Documentation) — Google Cloud
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission