Dataflow Architecture

1. Introduction

2. Why This Concept Exists

Traditional distributed computing frameworks (like early Apache Hadoop or Apache Spark versions) execute data sharding and state management directly on the worker nodes. This architecture creates several bottlenecks in production:

• Resource Contention: The same worker VMs must process records, perform disk-based shuffle operations, and keep track of aggregation window states, leading to CPU/RAM starvation.
• Slow Autoscaling: Adding or removing workers requires moving stored data shards between VMs, which is extremely slow and resource-intensive.
• Lack of Isolation: A memory leak or failure in the processing code can corrupt the internal state database.

Dataflow's decoupled architecture moves data shuffle and streaming state off the worker nodes into dedicated cloud-hosted services, ensuring workers remain stateless and highly responsive to scaling demands.

3. Key Terminology

• Control Plane: The Dataflow service backend that optimizes the job execution graph, directs workers, and monitors job health.
• Data Plane: The collection of Compute Engine VM workers executing user code.
• Dataflow Shuffle Service: A managed, off-worker service that handles the grouping and sorting of keys for batch pipelines.
• Streaming Engine: A managed, off-worker service that stores and updates state and windowing information for streaming pipelines.
• Dataflow Runner v2: The modern, unified execution backend that utilizes the Portability Framework to run pipelines in isolated container environments.

4. How It Works

Dataflow distributes tasks using three primary components:

1. Job Optimizer: When a job is submitted, the control plane optimizes the user's DAG (e.g., fusing adjacent transforms together to avoid network hops).
2. Externalized Data Plane:
   • For Batch pipelines, grouping steps (GroupByKey, CoGroupByKey) leverage the Dataflow Shuffle Service instead of writing files to local worker disks.
   • For Streaming pipelines, stateful operations use the Streaming Engine to save windowing state, watermarks, and user-defined state to an external high-throughput storage backend.
3. Stateless Workers: Because shuffle data and streaming states are externalized, Dataflow workers only perform CPU operations (processing, filtering, mapping). If a worker VM crashes or a new worker is added via autoscaling, no state data needs to be copied or redistributed over the network.

5. Visual Diagram

6. Code Example

The following pipeline options configuration shows how to enable the modern architecture options (Runner v2 and Streaming Engine) programmatically:

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

def run_architectural_pipeline():
    options = PipelineOptions()
    
    # 1. Configure standard GCP Options
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = "my-gcp-project"
    gcp_options.region = "us-central1"
    gcp_options.job_name = "architecture-demo-job"
    gcp_options.staging_location = "gs://my-bucket/staging"
    gcp_options.temp_location = "gs://my-bucket/temp"
    
    # 2. Set Runner to Dataflow
    options.view_as(StandardOptions).runner = "DataflowRunner"
    
    # 3. Enable Runner v2 & Streaming Engine explicitly
    # Note: Modern SDKs enable these by default for streaming jobs.
    options.view_as(StandardOptions).streaming = True
    
    # Set experiments for Runner v2 and Streaming Engine
    gcp_options.experiments = [
        "use_runner_v2",
        "enable_streaming_engine"
    ]
    
    with beam.Pipeline(options=options) as p:
        (p
         | "ReadStream" >> beam.Create([{"sensor": "temp-1", "val": 22.4}])
         | "Format" >> beam.Map(lambda data: f"Sensor: {data['sensor']} -> {data['val']}")
         | "Log" >> beam.Map(print)
        )

if __name__ == "__main__":
    run_architectural_pipeline()

7. Code Explanation

• StandardOptions.streaming = True switches the pipeline mode from Batch to Streaming.
• gcp_options.experiments = ["use_runner_v2", "enable_streaming_engine"] activates modern execution features. Runner v2 allows Python code to execute within standardized container environments (SDK Harness), separating customer code execution from the runner VM system processes.
• enable_streaming_engine shifts windowing and state storage from worker VM RAM/disk to Google Cloud's dedicated Streaming Engine backend, reducing VM load and improving performance.

8. Real Production Example

9. Common Mistakes

• Disabling Streaming Engine on Large Jobs: Not enabling Streaming Engine for high-throughput streaming jobs, which leads to high local disk usage and memory thrashing on workers.
• Assuming Shared Disk State: Writing intermediate states to local VM disks (/tmp/...) in custom DoFns, assuming other workers or subsequent steps can read them. Workers are stateless; data must only pass through PCollections.

10. Best Practices

• Always enable Runner v2 (use_runner_v2) to utilize the latest performance upgrades, container environments, and debugging features.
• Always enable Streaming Engine for streaming pipelines to decouple state storage, resulting in smoother autoscaling and cheaper VM costs.
• Maintain regional consistency: ensure your GCS bucket, Dataflow workers, and Streaming Engine end points are located in the same GCP region to prevent high egress costs.

11. Summary

• Dataflow decouples the Control Plane from the Data Plane.
• The Shuffle Service externalizes batch groupings to improve speed.
• Streaming Engine externalizes streaming states to improve scaling flexibility.
• Runner v2 provides containerized execution for better portability.

12. Interactive Challenges

13. Related Content

• Direct Runner vs Dataflow Runner
• Autoscaling
• Worker Configuration