Data Transformation

Project structure

Structure dbt projects in four layers. Keep each layer strictly separated: staging models only reference sources, intermediate models only reference staging, marts reference intermediate. Cross-layer references create hidden dependencies that break when the referenced model changes.

Incremental models

Configure every Silver and Gold model as incremental from day one. Full refresh is safe at small scale and catastrophically expensive at large scale. Do not wait until a model starts timing out to add incremental logic. Retrofitting it onto a model with downstream dependencies is significantly harder than building it in. Use unique_key with incremental_strategy = 'merge' to handle late-arriving updates and CDC corrections. Always include a lookback window wider than your expected late data window.

-- models/silver/orders.sql
{{
  config(
    materialized         = 'incremental',
    unique_key           = 'order_id',
    incremental_strategy = 'merge',
    partition_by = {
      'field':       'placed_date',
      'data_type':   'date',
      'granularity': 'day'
    },
    cluster_by = ['customer_id']
  )
}}

with source as (
  select
    order_id,
    customer_id,
    amount_gbp,
    placed_at,
    date(placed_at)  as placed_date,
    _ingested_at
  from {{ source('bronze', 'orders') }}

  {% if is_incremental() %}
  -- 3-hour lookback covers late-arriving events without a full-table scan.
  -- Adjust the window based on your domain's SLA for late data.
  where _ingested_at >= (
    select max(_ingested_at) - interval 3 hour
    from {{ this }}
  )
  {% endif %}
),

-- Deduplicate within the batch: keep the most recently ingested version per order.
-- This handles re-delivered events and out-of-order CDC rows.
deduplicated as (
  select *,
    row_number() over (
      partition by order_id
      order by _ingested_at desc
    ) as rn
  from source
)

select
  order_id,
  customer_id,
  amount_gbp,
  placed_at,
  placed_date,
  _ingested_at
from deduplicated
where rn = 1

dbt tests as the quality gate

dbt tests are the primary data quality gate for batch transforms. Configure them at the schema level so they run automatically after every model build. Failing tests should block promotion to Gold and page the domain owner, not just log a warning.

Slowly changing dimensions (SCD Type 2)

Use dbt snapshots for dimensions that change over time and where history matters: customer address, product price tier, organisational hierarchy. dbt snapshots apply the SCD Type 2 pattern automatically: add dbt_valid_from and dbt_valid_to columns, keep the full history, and expose a current-record view. Configure the snapshot strategy as check on the columns that define a change, not timestamp, since the timestamp strategy breaks when source systems backfill historical records.

dbt Cloud vs dbt Core

Use dbt Core with Cloud Composer (Airflow) if the organisation already operates Airflow and wants orchestration in one place. The Airflow dbt operator runs dbt commands as tasks with full dependency tracking. Use dbt Cloud if the data team wants a managed scheduler, a built-in IDE, and job history without operating Composer. dbt Cloud costs more but eliminates Composer overhead for teams that are only using Airflow for dbt scheduling. Do not use scheduled BigQuery queries or Dataform for complex multi-model dependency graphs because dbt's DAG-based dependency resolution is significantly more maintainable.

Beam programming model

A Beam pipeline is a DAG of transformations applied to PCollections. A PCollection is an immutable, potentially infinite dataset. The same abstraction works for a bounded batch file and an unbounded streaming topic. A PTransform is a function from PCollection to PCollection. Compose them with the pipe operator (|) to build the graph. The runner (Dataflow, DirectRunner, Flink) is pluggable at launch time.

Windowing, triggers, and watermarks

Streaming pipelines that aggregate over time require explicit window definitions. Watermarks tell Beam when to close a window and emit results: the watermark advances as event timestamps progress, and records arriving after the watermark advances are considered late. Configure allowed lateness for every windowed aggregation. Without it, late records are silently dropped.

Flex Templates

Package every Dataflow pipeline as a Flex Template: a containerised job definition stored in Artifact Registry with a metadata JSON file describing its parameters. Flex Templates decouple pipeline deployment from pipeline invocation. CI/CD builds the container and uploads the template; operators or Airflow launch jobs from the template without needing access to the pipeline source code. This is the correct way to deliver reusable ingestion pipelines to domain teams. They supply topic name, target table, and schema version as parameters; the platform team owns the execution logic.

Code: streaming pipeline with quarantine routing

Read from Pub/Sub, parse Avro, route valid records to Bronze and failures to quarantine. The tagged output pattern routes inside a single DoFn pass, with no second pipeline stage needed.

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.gcp.pubsub import ReadFromPubSub
from apache_beam.io.gcp.bigquery import WriteToBigQuery, BigQueryDisposition
import fastavro, io, datetime

# Load ORDER_AVRO_SCHEMA from your schema registry or a local .avsc file.
# fastavro.parse_schema(json.loads(registry_client.get_latest_schema("orders.placed.v1")))
ORDER_AVRO_SCHEMA = fastavro.parse_schema({ ... })  # abbreviated

BRONZE_SCHEMA = {
    "fields": [
        {"name": "order_id",     "type": "STRING",    "mode": "REQUIRED"},
        {"name": "customer_id",  "type": "STRING",    "mode": "REQUIRED"},
        {"name": "amount_gbp",   "type": "FLOAT64",   "mode": "REQUIRED"},
        {"name": "placed_at",    "type": "TIMESTAMP", "mode": "REQUIRED"},
        {"name": "_ingested_at", "type": "TIMESTAMP", "mode": "REQUIRED"},
    ]
}

class ParseOrderEvent(beam.DoFn):
    """Parse Avro bytes. Route valid records and failures to separate output tags."""
    def process(self, element):
        try:
            record = fastavro.schemaless_reader(io.BytesIO(element), ORDER_AVRO_SCHEMA)
            yield beam.pvalue.TaggedOutput("valid", {
                "order_id":     record["order_id"],
                "customer_id":  record["customer_id"],
                "amount_gbp":   record["amount_cents"] / 100.0,
                "placed_at":    record["placed_at_ms"] / 1000.0,
                "_ingested_at": datetime.datetime.utcnow().isoformat(),
            })
        except Exception as exc:
            # Never silently drop. Route to quarantine with full error context.
            yield beam.pvalue.TaggedOutput("failed", {
                "raw_hex":   element.hex(),
                "error":     str(exc),
                "failed_at": datetime.datetime.utcnow().isoformat(),
            })

def run():
    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        results = (
            p
            | "Read Pub/Sub" >> ReadFromPubSub(
                subscription="projects/my-project/subscriptions/orders.placed.v1-bronze"
            )
            | "Parse Avro" >> beam.ParDo(ParseOrderEvent()).with_outputs("valid", "failed")
        )

        # Valid records → Bronze table. STREAMING_INSERTS deduplicates on insertId.
        (
            results.valid
            | "Write to Bronze" >> WriteToBigQuery(
                "my-project:bronze.orders",
                schema=BRONZE_SCHEMA,
                write_disposition=BigQueryDisposition.WRITE_APPEND,
                create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                method="STREAMING_INSERTS",
            )
        )

        # Failed records → quarantine table. Full error context retained for investigation.
        (
            results.failed
            | "Write to Quarantine" >> WriteToBigQuery(
                "my-project:bronze.orders_quarantine",
                schema={"fields": [
                    {"name": "raw_hex",   "type": "STRING"},
                    {"name": "error",     "type": "STRING"},
                    {"name": "failed_at", "type": "TIMESTAMP"},
                ]},
                write_disposition=BigQueryDisposition.WRITE_APPEND,
            )
        )

if __name__ == "__main__":
    run()

Flink vs Dataflow: when each wins

Stateful stream processing

Stateful transforms maintain per-key state across messages: computing a running total per customer, detecting a sequence of events, or deduplicating within a sliding window. In Beam, declare state using BagState, ValueState, or MapState annotations on a DoFn. State is partitioned by key and lives in the runner's state backend (Dataflow uses Bigtable-backed persistent state for large state). Use event-time timers to flush state when the watermark advances past a deadline, not on wall-clock time.

Stream-to-stream joins

Joining two streams requires both streams to have a common watermark strategy. If one stream is consistently faster than the other, the slower stream's watermark lags, holding the join window open and accumulating state. Set an explicit allowed lateness on both sides and accept that records arriving after the cutoff will be processed as late data. For high-cardinality joins (e.g., enriching order events with product catalog data), prefer a side input pattern over a stream join: load the reference data into memory on a refresh schedule and look up during the main stream's ParDo.

Partitioning and clustering as performance levers

Partition every Silver and Gold table by the date column most used in WHERE clauses, typically event date or processing date. Cluster by the columns most used in GROUP BY or JOIN ON clauses. Partitioning eliminates full-table scans; clustering reduces the data scanned within a partition. Both are defined at table creation time in dbt config or Terraform and cannot be changed without recreation. Get them right before the table accumulates data.

MERGE for incremental loads

When running incremental loads outside of dbt, use BigQuery MERGE. The pattern is identical to the dbt incremental model: source is a filtered slice of Bronze with a lookback window and deduplication applied; the MERGE statement upserts into Silver on the primary key.

-- Incremental load via MERGE (raw SQL, for Dataform or Cloud Composer scheduling)
-- Use this pattern when dbt is not managing the pipeline.

MERGE silver.orders AS target
USING (
  SELECT
    order_id,
    customer_id,
    amount_gbp,
    placed_at,
    DATE(placed_at)  AS placed_date,
    _ingested_at
  FROM bronze.orders
  -- 3-hour lookback covers late arrivals without scanning the full Bronze table.
  WHERE _ingested_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 3 HOUR)
  -- Deduplicate: one row per order_id, latest ingestion wins.
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY order_id ORDER BY _ingested_at DESC
  ) = 1
) AS source
ON target.order_id = source.order_id

WHEN MATCHED THEN UPDATE SET
  amount_gbp   = source.amount_gbp,
  placed_at    = source.placed_at,
  _ingested_at = source._ingested_at

WHEN NOT MATCHED THEN INSERT (
  order_id, customer_id, amount_gbp, placed_at, placed_date, _ingested_at
) VALUES (
  source.order_id, source.customer_id, source.amount_gbp,
  source.placed_at, source.placed_date, source._ingested_at
);

Materialized views for the serving layer

Create BigQuery materialized views on top of Gold tables for high-traffic BI dashboards. Materialized views are pre-computed and cached; BigQuery refreshes them automatically when the base table changes. They eliminate repeated full-table scans on large aggregation queries. Use them for the top-level metrics that Looker dashboards query on every page load, not for exploration queries that change shape frequently.

Strategies

Operational decision: defining the SLA

Define three thresholds for each domain. Records within the on-time window (e.g., under 15 minutes) are processed normally by the incremental pipeline. Records within the late window (e.g., 15 minutes to 24 hours) are caught by the lookback and processed on the next run, with no action required. Records outside the correction window (e.g., beyond 24 hours) require a targeted backfill, trigger an alert, and are logged in the incident register. Document these thresholds in the data product specification. Consumers need to know the maximum correction lag when building business reports.