Dataflow Runner (Introduction)

1. Introduction

2. Why This Concept Exists

Managing distributed processing clusters (like Spark or Flink) requires substantial operational effort. Teams must configure virtual machines, manage network connections, set up security policies, and scale nodes manually when data volume spikes.

3. Key Terminology

• Managed Service: A service where the cloud provider manages server setup, maintenance, scaling, and patching.
• Autoscaling: Automatically increasing or decreasing the number of VM workers based on processing load.
• Staging Location: A Google Cloud Storage (GCS) folder where the runner uploads the serialized pipeline files and dependencies.
• Temp Location: A GCS path used for storing temporary files generated during pipeline execution steps.

4. How It Works

• Local Submission: The developer runs the Python script locally, specifying --runner=DataflowRunner and GCP options.
• Job Creation: The Beam SDK compiles the DAG proto and sends it to the Dataflow Service API.
• Resource Provisioning: Dataflow starts Google Compute Engine worker VMs in the specified region.
• Distribution & Scaling: The service distributes data partitions across VMs, dynamically scaling workers as needed.
• Tear Down: Once complete, output data is written to sinks (e.g. BigQuery), and Dataflow terminates the VMs.

5. Visual Diagram

6. Code Example

pythonCopy
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions
import apache_beam as beam

def run_dataflow():
    # 1. Initialize Pipeline Options
    options = PipelineOptions()
    
    # 2. Configure GCP Settings
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = "my-gcp-project-id"
    gcp_options.region = "us-central1"
    gcp_options.job_name = "dataflow-intro-example"
    gcp_options.staging_location = "gs://my-bucket/staging"
    gcp_options.temp_location = "gs://my-bucket/temp"
    
    # 3. Set Execution Runner to Dataflow
    options.view_as(StandardOptions).runner = "DataflowRunner"
    
    # 4. Construct and Submit Pipeline
    with beam.Pipeline(options=options) as p:
        (p
         | "ReadWords" >> beam.Create(["cloud", "dataflow", "runner", "serverless"])
         | "UpperWords" >> beam.Map(lambda x: x.upper())
         | "LogOutput" >> beam.Map(print)) # In Dataflow, this prints to worker logs

if __name__ == "__main__":
    run_dataflow()

7. Code Explanation

• GoogleCloudOptions provides a typed interface for GCP configurations.
• staging_location and temp_location specify GCS buckets (gs://...). The local SDK packages code packages and uploads them to the staging path.
• runner = "DataflowRunner" tells Beam to compile and deploy the job to Google Cloud.
• beam.Map(print) outputs logs to the worker standard outputs. When running on Dataflow, these logs are captured by GCP Cloud Logging and visible in the Dataflow console.

8. Real Production Example

A media streaming service processes user interaction click events. They set up a streaming Dataflow job that reads click events from a Pub/Sub topic, parses and enriches the data, and writes the structured records directly to BigQuery. During high-traffic sporting events, Dataflow automatically scales workers from 2 to 50 VMs, returning to 2 VMs when traffic subsides.

9. Common Mistakes

• Undefined Staging or Temp Locations: Forgetting to configure staging_location or temp_location. Without these GCS buckets, the SDK cannot upload code payloads, and the job submission will fail.
• Missing API Enablement: Attempting to run a job without enabling the Dataflow API and Compute Engine API in your GCP console.

10. Best Practices

• Ensure that the service account executing the job has permission to read/write to the staging/temp GCS buckets.
• Always specify a geographic region (e.g. us-central1) for worker VMs to comply with data residency and minimize network costs.

11. Summary

• Dataflow is a fully-managed serverless runner on Google Cloud.
• It automates VM provisioning, scaling, and tear down.
• Requires setting GCS staging and temp paths in options.
• Autoscales dynamically using liquid sharding.
• Logs are centralized in GCP Cloud Logging.

12. Interactive Challenges

13. Related Content

• Pipeline Runners
• Running Your First Pipeline
• Direct Runner