Ingesting and Processing Data

Sources and Sinks

Every data pipeline has sources (where data comes from) and sinks (where it goes). For the exam, know which GCP services serve as common sources and sinks:

Encryption in Transit and at Rest

All data moving between GCP services is encrypted in transit by default using TLS. For data flowing between on-premises and GCP, use Cloud VPN or Cloud Interconnect for encrypted private connectivity.

The Beam Programming Model

Apache Beam provides a unified model for both batch and streaming. The key abstractions are:

Core Transforms

Complete Pipeline Example

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

class ParseEvent(beam.DoFn):
    """Parse JSON event and add derived fields."""
    def process(self, element):
        import json
        record = json.loads(element)
        record['event_hour'] = record['timestamp'][:13]
        yield record

class FilterValid(beam.DoFn):
    """Filter out invalid records."""
    def process(self, element):
        if element.get('user_id') and element.get('amount', 0) >= 0:
            yield beam.pvalue.TaggedOutput('valid', element)
        else:
            yield beam.pvalue.TaggedOutput('invalid', element)

options = PipelineOptions([
    '--runner=DataflowRunner',
    '--project=my-project',
    '--region=us-central1',
    '--temp_location=gs://my-bucket/temp',
    '--staging_location=gs://my-bucket/staging',
    '--streaming',                      # Enable streaming mode
    '--autoscaling_algorithm=THROUGHPUT_BASED',
    '--max_num_workers=20',
    '--usePublicIps=false',             # Private workers
])

with beam.Pipeline(options=options) as p:
    # Read from Pub/Sub
    events = (
        p
        | 'ReadPubSub' >> beam.io.ReadFromPubSub(
            topic='projects/my-project/topics/events')
        | 'Decode' >> beam.Map(lambda x: x.decode('utf-8'))
        | 'Parse' >> beam.ParDo(ParseEvent())
    )

    # Split into valid and invalid
    tagged = events | 'Validate' >> beam.ParDo(FilterValid()).with_outputs('valid', 'invalid')

    # Write valid events to BigQuery
    tagged.valid | 'WriteBQ' >> beam.io.WriteToBigQuery(
        table='my-project:dataset.events',
        schema='user_id:STRING,amount:FLOAT,timestamp:TIMESTAMP,event_hour:STRING',
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        insert_retry_strategy=beam.io.gcp.bigquery_tools.RetryStrategy.RETRY_ON_TRANSIENT_ERROR,
        method=beam.io.WriteToBigQuery.Method.STREAMING_INSERTS,
    )

    # Write invalid to dead-letter table
    tagged.invalid | 'WriteErrors' >> beam.io.WriteToBigQuery(
        table='my-project:dataset.events_errors',
        schema='user_id:STRING,amount:FLOAT,timestamp:STRING',
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
    )

Cluster Mode vs Serverless

# Create an ephemeral Dataproc cluster, run a job, delete it
gcloud dataproc clusters create my-cluster \
  --region=us-central1 \
  --num-workers=4 \
  --worker-machine-type=n2-standard-8 \
  --num-secondary-workers=8 \
  --secondary-worker-type=preemptible \
  --autoscaling-policy=my-policy \
  --enable-component-gateway \
  --optional-components=JUPYTER,HIVE_WEBHCAT \
  --initialization-actions=gs://my-bucket/init/install-deps.sh \
  --max-idle=30m          # Auto-delete after 30 min idle

# Submit a PySpark job
gcloud dataproc jobs submit pyspark \
  --cluster=my-cluster \
  --region=us-central1 \
  --py-files=gs://my-bucket/libs/utils.py \
  gs://my-bucket/jobs/transform.py \
  -- --input=gs://raw-data/ --output=gs://processed/

# Dataproc Serverless (no cluster to manage)
gcloud dataproc batches submit pyspark \
  --region=us-central1 \
  --deps-bucket=gs://my-bucket/deps \
  gs://my-bucket/jobs/transform.py \
  -- --input=gs://raw-data/ --output=gs://processed/

When to Use Spark vs Beam (Dataflow)

Pub/Sub Deep Dive

Cloud Pub/Sub is a global, real-time messaging service. Key concepts for the exam:

# Create a topic with a schema
gcloud pubsub topics create events \
  --schema=projects/my-project/schemas/event-schema \
  --message-encoding=JSON

# Create a subscription with dead-letter topic and ordering
gcloud pubsub subscriptions create events-sub \
  --topic=events \
  --dead-letter-topic=projects/my-project/topics/events-dlq \
  --max-delivery-attempts=10 \
  --enable-message-ordering \
  --ack-deadline=60

# Create a BigQuery subscription (direct to BQ, no pipeline needed)
gcloud pubsub subscriptions create events-bq-sub \
  --topic=events \
  --bigquery-table=my-project:dataset.events \
  --use-topic-schema \
  --write-metadata

Windowing Strategies

Windows group unbounded data into finite chunks for aggregation. The four window types in Beam:

import apache_beam as beam
from apache_beam import window

# Fixed windows: 5-minute tumbling windows
events | 'FixedWindow' >> beam.WindowInto(window.FixedWindows(5 * 60))

# Sliding windows: 10-minute window, sliding every 2 minutes
events | 'SlidingWindow' >> beam.WindowInto(window.SlidingWindows(10 * 60, 2 * 60))

# Session windows: gap of 30 minutes
events | 'SessionWindow' >> beam.WindowInto(window.Sessions(30 * 60))

# After windowing, aggregate per window
(events
 | 'FixedWindow' >> beam.WindowInto(window.FixedWindows(60))  # 1-min windows
 | 'AddKey' >> beam.Map(lambda e: (e['sensor_id'], e['reading']))
 | 'AvgPerWindow' >> beam.CombinePerKey(beam.combiners.MeanCombineFn())
 | 'WriteBQ' >> beam.io.WriteToBigQuery(...)
)

Late Data, Watermarks, and Triggers

In streaming systems, data can arrive late (after the window has "closed"). Beam uses three mechanisms to handle this:

• Watermark — An estimate of event-time progress. When the watermark passes the end of a window, Beam considers the window "complete." Dataflow tracks watermarks automatically.
• Allowed lateness — How long after the watermark passes to still accept late data. Late elements are added to their original window and trigger re-computation.
• Triggers — Control when window results are emitted. AfterWatermark (default), AfterProcessingTime, AfterCount, or Repeatedly for periodic updates.

from apache_beam.transforms.trigger import (
    AfterWatermark, AfterProcessingTime, AccumulationMode, AfterCount
)

# Window with late data handling
events | beam.WindowInto(
    window.FixedWindows(5 * 60),          # 5-minute windows
    trigger=AfterWatermark(
        early=AfterProcessingTime(60),     # Speculative results every 60s
        late=AfterCount(1),                # Fire for each late element
    ),
    allowed_lateness=Duration(seconds=3600),  # Accept up to 1 hour late
    accumulation_mode=AccumulationMode.ACCUMULATING,  # Include all data
)

BigQuery Streaming Options

Cloud Composer DAGs

Cloud Composer is managed Apache Airflow for orchestrating data pipelines. Key exam concepts:

• DAG (Directed Acyclic Graph) — Defines task dependencies and execution order
• Operators — Define what each task does (BigQueryInsertJobOperator, DataflowCreatePythonJobOperator, GCSToGCSOperator, etc.)
• Sensors — Wait for external conditions (GCSObjectExistenceSensor, BigQueryTableExistenceSensor)
• XCom — Pass data between tasks (small metadata only, not large datasets)
• Composer 2 — Auto-scaling workers, smaller environment footprint, lower cost

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryInsertJobOperator,
    BigQueryCheckOperator,
)
from airflow.providers.google.cloud.operators.dataflow import (
    DataflowCreatePythonJobOperator,
)
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor

default_args = {
    'owner': 'data-eng',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['alerts@example.com'],
}

with DAG(
    dag_id='daily_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=days_ago(1),
    catchup=False,
    tags=['production', 'bigquery'],
) as dag:

    # Wait for source file
    wait_for_file = GCSObjectExistenceSensor(
        task_id='wait_for_source',
        bucket='raw-data',
        object='events/{{ ds }}/data.parquet',
        timeout=3600,
        poke_interval=300,
    )

    # Run Dataflow pipeline
    run_pipeline = DataflowCreatePythonJobOperator(
        task_id='run_dataflow',
        py_file='gs://my-bucket/pipelines/transform.py',
        job_name='daily-transform-{{ ds_nodash }}',
        options={
            'input': 'gs://raw-data/events/{{ ds }}/',
            'output': 'my-project:dataset.events',
            'temp_location': 'gs://my-bucket/temp',
        },
        project_id='my-project',
        location='us-central1',
    )

    # Data quality check
    quality_check = BigQueryCheckOperator(
        task_id='quality_check',
        sql="""
            SELECT COUNT(*) > 0
            FROM `my-project.dataset.events`
            WHERE DATE(event_time) = '{{ ds }}'
        """,
        use_legacy_sql=False,
    )

    # Build aggregate table
    build_agg = BigQueryInsertJobOperator(
        task_id='build_aggregates',
        configuration={
            'query': {
                'query': """
                    CREATE OR REPLACE TABLE `my-project.dataset.daily_agg`
                    PARTITION BY event_date
                    AS SELECT
                        DATE(event_time) AS event_date,
                        user_id,
                        COUNT(*) AS event_count,
                        SUM(amount) AS total_amount
                    FROM `my-project.dataset.events`
                    GROUP BY 1, 2
                """,
                'useLegacySql': False,
            }
        },
    )

    wait_for_file >> run_pipeline >> quality_check >> build_agg

Google Cloud Workflows

For simpler orchestration without the overhead of Airflow, Workflows is a serverless orchestration service. It uses YAML/JSON to define steps that call GCP APIs, Cloud Functions, or Cloud Run services.

CI/CD for Data Pipelines

Production pipelines should follow CI/CD practices:

• Source control — Pipeline code (Beam, Spark, DAGs) in Git
• Unit tests — Test DoFn logic, SQL queries with mock data
• Integration tests — Run with DirectRunner on sample data
• Deployment — Cloud Build triggers deploy DAGs to Composer, Dataflow templates to GCS
• Dataflow Flex Templates — Package pipeline as a Docker container for versioned, parameterized deployments