Pipeline Basics

1. Introduction

2. Why This Concept Exists

3. Key Terminology

• Pipeline Object: The main container that manages all the steps and connections of your data flow.
• DAG (Directed Acyclic Graph): A map of your pipeline steps showing the direction data moves. Acyclic means it flows forward and never loops backward.
• PipelineOptions: A configuration object containing settings like which runner to use (e.g. Flink or Dataflow) and how many worker machines to start.

4. How It Works

1. Configure: Create a PipelineOptions instance to set runtime options.
2. Initialize: Create a pipeline container: with beam.Pipeline(options=options) as p:.
3. Apply transforms: Use the pipe operator | to chain operations.
4. Label: Label each step using the >> syntax (e.g., "FilterLogs" >> beam.Filter(...)).
5. Execute: When you exit the with block, the pipeline automatically executes.

5. Visual Diagram

Here is how data flows sequentially inside a basic pipeline:

6. Code Example

Here is a complete, runnable script demonstrating a basic pipeline setup:

pythonCopy
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# 1. Define configuration settings
options = PipelineOptions()

# 2. Open the pipeline context
with beam.Pipeline(options=options) as p:
    # 3. Read, transform, and write data
    (p
     | "LoadNames" >> beam.Create(["alice", "bob", "charlie"])
     | "Capitalize" >> beam.Map(lambda name: name.capitalize())
     | "Print" >> beam.Map(print))

7. Code Explanation

• PipelineOptions() initializes default configuration parameters.
• with beam.Pipeline(options=options) as p: handles starting and stopping execution hooks.
• p | "LoadNames" >> beam.Create(...) inputs static values.
• | "Capitalize" >> beam.Map(...) applies capitalization.
• | "Print" >> beam.Map(print) prints the results.

8. Real Production Example

In production, a pipeline is configured to run on Google Cloud Dataflow:

pythonCopy
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions(
    runner="DataflowRunner",
    project="my-gcp-project",
    region="us-central1",
    temp_location="gs://my-bucket/temp"
)

This instructs the pipeline to run on Google Cloud workers instead of your local machine.

9. Common Mistakes

• Re-using the same name label: Every label string (e.g. "LoadNames") must be unique. Duplicate labels cause execution crashes.
• Running without context blocks: If you construct a pipeline without a with statement, it will not run unless you explicitly call p.run() at the end of your script.

10. Best Practices

• Always provide unique names for transforms using the >> operator.
• Keep pipeline construction code separated from your business transformation logic functions for easier unit testing.

11. Summary

• A pipeline is a Directed Acyclic Graph (DAG).
• Chained using the | operator and labeled using >>.
• Executes automatically upon exiting a with beam.Pipeline() block.

12. Interactive Challenges

13. Related Content

• Pipeline Options
• PCollection Basics