Nova Retail: Dockerized data platform
with Airflow-orchestrated ETL

Five containers, one pipeline, two operating modes.

The entire platform runs as a five-service Docker Compose stack. PostgreSQL 15 as the warehouse, a custom Spark container as the execution environment, and Apache Airflow's three components (init, webserver, scheduler) as the orchestration layer. Persistent named volumes keep data across restarts. A health check on PostgreSQL gates dependent services from starting before the database is genuinely ready. One docker-compose up brings the whole stack online.

The Spark container is "sidecar Spark": it runs sleep infinity and stays alive as a long-lived execution environment. Airflow doesn't run Spark itself. When the DAG triggers, the BashOperator calls docker exec spark /opt/spark/bin/spark-submit ... to inject jobs into the running Spark container. This keeps orchestration decoupled from execution, lets Airflow restart independently of Spark, and matches the topology used when deploying against a managed cluster (Databricks, EMR, Dataproc) where docker exec becomes a cluster-submit API call.

Parameterised DAG: same code, two modes

The Airflow DAG accepts two parameters via Airflow's native Param API: scale_data (boolean, defaults to false) and scale_multiplier (integer 1-5, defaults to 2). Scheduled runs use the defaults, producing analytical truth from real Olist data. Manual triggers via the Airflow UI surface a form letting the operator request a scaled run for architecture validation. Both flow through the same DAG, the same Spark container, the same pipeline code. Only the environment variables passed through docker exec change.

with DAG(
    dag_id="retail_etl_pipeline",
    schedule_interval="0 6 * * *",  # daily at 06:00 UTC
    params={
        "scale_data": Param(False, type="boolean"),
        "scale_multiplier": Param(2, type="integer", minimum=1, maximum=5),
    },
) as dag:

    run_spark_etl = BashOperator(
        task_id="run_full_etl_pipeline",
        bash_command=(
            "docker exec "
            "-e SCALE_DATA={{ 'true' if params.scale_data else 'false' }} "
            "-e SCALE_MULTIPLIER={{ params.scale_multiplier }} "
            "spark /opt/spark/bin/spark-submit "
            "--master local[*] "
            "/opt/retail_project/etl/run_pipeline.py"
        ),
    )

The Jinja templating ({{ params.X }}) interpolates parameters into the docker exec -e flags at trigger time. The pipeline reads them as environment variables and adjusts behaviour accordingly.

Three idempotency layers, all overwrite

Bronze writes raw Parquet with mode="overwrite". Silver writes joined Parquet with mode="overwrite". Gold loads to PostgreSQL with mode="overwrite". The honest tradeoff: simple, deterministic, idempotent. Running the pipeline twice in a row produces the same warehouse state. Re-running after a failure picks up cleanly. The cost is that every run rebuilds every layer; for the data volumes Nova Retail operates at, that cost is negligible compared to the simplicity it buys.

Partition the facts. Cache the joins. Validate every load.

Three engineering patterns carry most of the scalability weight in this pipeline. Plus one pattern that catches silent failures most pipelines never notice.

1. Year/month partitioning on fact-shaped Bronze tables

The Bronze layer writes orders and order_items with .partitionBy("order_year", "order_month"), deriving partition columns from order_purchase_timestamp and shipping_limit_date respectively. This isn't a transformation of the data; the source columns are preserved. The partition columns are storage-layout metadata that enable partition pruning on read. Small dimension tables (customers, sellers, products) write unpartitioned. At current 555K-row scale this is invisible. At any volume beyond that, it's the single biggest architectural lever.

PARTITIONED_DATASETS = {
    "orders":      "order_purchase_timestamp",
    "order_items": "shipping_limit_date",
}

def bronze_layer(datasets):
    for name, df in datasets.items():
        output_path = layer_path("bronze", name)

        if name in PARTITIONED_DATASETS:
            ts_col = PARTITIONED_DATASETS[name]
            df_partitioned = (
                df.withColumn("order_year", year(col(ts_col)))
                  .withColumn("order_month", month(col(ts_col)))
            )
            df_partitioned.write \
                .mode("overwrite") \
                .partitionBy("order_year", "order_month") \
                .parquet(output_path)
        else:
            df.write.mode("overwrite").parquet(output_path)

2. Broadcast joins on small dimensions

The Silver layer joins seven tables. Two of them are small enough to broadcast to every Spark executor instead of shuffling. sellers at 3K rows and products at 33K rows are stable across all tested scales. Explicit broadcast() hints make this deterministic instead of relying on Spark's auto-broadcast threshold. The customers table was tested as a broadcast target and removed when scaled testing showed it could exceed safe broadcast size; the discipline is to broadcast where the small side stays small, shuffle where it might grow.

silver_df = (
    orders
    .join(customers, "customer_id", "left")
    .join(payments, "order_id", "left")
    .join(items, "order_id", "left")
    .join(broadcast(products), "product_id", "left")   # 33K rows, safe
    .join(broadcast(sellers), "seller_id", "left")     # 3K rows, safe

    # Numeric precision, date parts, then filter
    .withColumn("payment_value", col("payment_value").cast("DECIMAL(10,2)"))
    .withColumn("order_year", year(col("order_purchase_date")))
    .filter(col("payment_value").isNotNull())
)

# Cache before reuse: count + write + downstream Gold all hit this frame
silver_df = silver_df.cache()

3. Explicit caching on the join chain

The Silver layer's joined output gets cached before any action. The Gold layer caches fact_sales after the column selection because it's used four times downstream (count + write + two summary aggregations). Without caching, Spark may recompute these chains on each subsequent action. With caching, the join work happens once and the result is held in memory (with disk spillover via MEMORY_AND_DISK semantics) for everything downstream. This is the optimisation that made Gold at 2x scale finish in 59 seconds instead of several minutes.

4. Row-count validation on every Postgres load

After each PostgreSQL write, the loader reads back the target table via JDBC and counts rows. If the count matches the source DataFrame, validation passes. If not, a warning is logged. This catches silent failures: writes that "succeed" at the protocol level but lose rows due to constraint violations, transaction rollbacks, or driver bugs. Three Gold tables, three validation passes per pipeline run.

def load_df_to_postgres(df, schema_name, table_name, mode="overwrite"):
    row_count = df.count()

    df.write \
      .format("jdbc") \
      .option("url", jdbc_url) \
      .option("dbtable", f"{schema_name}.{table_name}") \
      .option("batchsize", 10000) \
      .mode(mode) \
      .save()

    # Validation: read back, count, compare
    check_df = df.sparkSession.read \
        .format("jdbc") \
        .option("url", jdbc_url) \
        .option("dbtable", f"{schema_name}.{table_name}") \
        .load()

    loaded = check_df.count()
    if loaded == row_count:
        logging.info("✅ Validation OK — row count matches")
    else:
        logging.warning(f"⚠️ Mismatch! Source: {row_count} | DB: {loaded}")

Observability via shared logger

The same logger.py from XTD and FibbieBanks powers Nova's observability: @timed decorator for stage durations, cross-platform UTF-8 detection for emoji-safe console output, dual-handler setup (colourised console + rotating UTF-8 file), section() banners between stages. One pipeline.log file tells the whole story of any run.

2026-06-08 13:56:21 INFO 🔷 Creating Spark Session
2026-06-08 13:57:02 INFO ============== Spark Session created successfully ==============
2026-06-08 13:57:02 INFO 🔷 STEP 1: EXTRACT
2026-06-08 13:57:30 INFO orders: 99441 rows · customers: 99441 · order_items: 112650
                                       order_payments: 103886 · reviews: 104162 · products: 32951 · sellers: 3095

2026-06-08 13:57:30 INFO 🔷 STEP 2: BRONZE
2026-06-08 13:58:03 INFO Successfully written to Bronze: orders (partitioned by year/month)
2026-06-08 13:58:03 INFO Successfully written to Bronze: order_items (partitioned by year/month)
2026-06-08 13:58:03 INFO Successfully written to Bronze: customers · sellers · products · order_payments · reviews

2026-06-08 13:58:03 INFO 🔷 STEP 3: SILVER
2026-06-08 13:58:37 INFO Valid records after cleaning: 118,431

2026-06-08 13:58:37 INFO 🔷 STEP 4: GOLD
2026-06-08 13:59:14 INFO Fact Sales created: 118,431 records
2026-06-08 13:59:14 INFO Sales Summary created: 1,394 groups
2026-06-08 13:59:14 INFO Sales by Month + State created: 565 groups

2026-06-08 13:59:14 INFO 🔷 STEP 5: LOAD TO POSTGRESQL
2026-06-08 13:59:43 INFO ✅ Load successful → retail_gold.fact_sales — 118,431 rows
2026-06-08 13:59:43 INFO ✅ Validation OK — row count matches
2026-06-08 13:59:53 INFO ✅ Load successful → retail_gold.sales_summary — 1,394 rows
2026-06-08 13:59:53 INFO ✅ Validation OK — row count matches
2026-06-08 14:00:05 INFO ✅ Load successful → retail_gold.sales_by_month_state — 565 rows
2026-06-08 14:00:05 INFO ✅ Validation OK — row count matches

2026-06-08 14:00:06 INFO ✅✅✅ PIPELINE COMPLETED ✅✅✅
2026-06-08 14:00:07 INFO ⏱️ Step completed in 3m 11s

What this pipeline produces.

Real numbers from two captured pipeline logs (June 2026): the stock production run that anchors analytical truth, and the 2x scaled validation run that proves the architecture handles substantially larger volumes than current production scale.

Stock run · runtime breakdown

2x architecture validation · what it proves

The 2x scaled run replicates source data via union, producing ~1.1M source rows that join-multiply to 7,533,104 Silver records. The pipeline processed all 7.5M rows end-to-end in 13m 14s with row-count validation passing on every Postgres load. Honest framing: this validates throughput, not analytical correctness. Row replication produces duplicate keys, which compound through joins; the 7.5M figure represents pipeline row-handling capacity, not distinct logical sales.

What it does prove, concretely: PySpark's distributed transform engine handled 7.5M rows through the medallion without intervention. The caching strategy paid off (Gold built three tables from a 7.5M-row cached fact_sales in 59 seconds total). Bronze partitioning scaled linearly (2x volume produced ~2x Bronze write time, not exponential). The JDBC load completed (7.5M rows over a single connection in ~5 minutes write + ~3.5 minutes validation). Three Gold tables loaded and validated successfully.