Multi-Stage ETL — chained two-hop JOIN + DST timezone (capstone)

View on GitHub

What

A transitive (two-hop) JOIN that no single pass can do — order → product → category → name — chained across passes, while normalising three different date formats and bridging CSV ↔ JSON, finishing with a DST-aware Europe/Prague timestamp.

Synthetic / teaching example

The data here is constructed, not sourced — orders.csv (US MM/DD/YYYY), products.csv (EU DD.MM.YYYY), categories.in.json (long YYYYMMDD). The problem class — snowflake relationships, format drift across sources, and timezone correctness — is universal; the rows are not.

Why interesting

Two limits force the chaining: bxp reads one file per pass, and runs one pre_pass per pass. The second hop's key (category_id) does not exist on an order until the first hop has run, so it cannot be a single lookup — you combine, join, combine again, join again. On top of that, the author knows each source's date convention up front, and the output target is Prague local time with a correct summer/winter offset — which TZ_OFFSET resolves from the bundled IANA tz database in a single call.

flowchart TD
    O["orders.csv<br/><small>US MM/DD/YYYY</small>"] --> C1["1 · combine_op"]
    P["products.csv<br/><small>EU DD.MM.YYYY</small>"] --> C1
    C1 --> J1["2 · join_product — hop 1<br/><small>+category_id · dates → ISO</small>"]
    J1 --> C2["3 · combine_oc<br/><small>+ DST timestamp</small>"]
    CAT["categories.in.json<br/><small>long YYYYMMDD</small>"] --> C2
    C2 --> J2["4 · join_category — hop 2<br/><small>+category_name · created → ISO</small>"]
    J2 --> R["1-final.json"]

Problem class documented in. (sources for the problem class — not for the data)

• Snowflake-schema multi-hop joins are standard dimensional modelling (Kimball).
• EU DST: clocks switch on the last Sunday of March / October (Directive 2000/84/EC) — the rule TZ_OFFSET applies for Europe/Prague.

The trick — four passes, and what each one leaves behind

Run all four at once with bxp-cli --config ./sample.json (the template is heavily commented — open the sample.json tab at the bottom). Each pass writes a real file, and the point of the pipeline is easiest to see by reading those four files in order. Every one of them is committed and pinned by a golden, so what you see below is exactly what the run produces.

Each date is normalised in the pass where its source format is known, and the timestamp is derived only once order_date is ISO — so the final join pass stays free of date math.

Pass 1 · combine_op — stack the two CSVs

combined_output writes orders and products into one file, and _type is derived from which key a row has (orders have order_id, products do not). A pre_pass reads one file per pass, so the two tables have to share a file before they can be joined at all.

Nothing is converted yet — both source date formats are still verbatim, US 01/15/2024 next to EU 02.01.2023:

1-combine_op-combined.csvx

_type,order_id,product_id,order_date,category_id,added_date
order,1001,P-9,01/15/2024,,
order,1002,P-7,07/20/2024,,
order,1003,P-9,11/02/2024,,
product,,P-9,,C-3,02.01.2023
product,,P-7,,C-1,15.06.2023

Pass 2 · join_product — hop 1, order → product

The pre_pass indexes the product rows by product_id; every order row then LOOKUPs its category_id out of that index. row_rules emits only the orders — the product rows were scaffolding for the lookup, and they disappear here.

This is also where both CSV date formats become ISO, each with the format its own source uses. Output is JSON so the next pass can stack it with the categories file:

1-enriched.in.json

[
{"order_id":"1001","product_id":"P-9","order_date":"2024-01-15","category_id":"C-3","product_added":"2023-01-02"},
{"order_id":"1002","product_id":"P-7","order_date":"2024-07-20","category_id":"C-1","product_added":"2023-06-15"},
{"order_id":"1003","product_id":"P-9","order_date":"2024-11-02","category_id":"C-3","product_added":"2023-01-02"}
]

Note what an order now knows that it could not know before: category_id. That is the key the second hop needs, and it did not exist on an order until this pass ran — which is the whole reason one pre_pass cannot do the job.

Pass 3 · combine_oc — stack in the categories, stamp the timestamp

The enriched orders and the categories file are both JSON, so combined_output merges them the same way pass 1 merged the CSVs, _type and all.

The DST-aware timestamp is built here, not in the final pass, because this is where order_date is already ISO and still present as a field. Watch the offset follow the season — +01:00 in January, +02:00 in July, back to +01:00 in November:

1-combine_oc-combined.csvx

_type,order_id,product_id,order_ts,category_id,product_added,category_name,created
order,1001,P-9,2024-01-15T00:00:00+01:00,C-3,2023-01-02,,
order,1002,P-7,2024-07-20T00:00:00+02:00,C-1,2023-06-15,,
order,1003,P-9,2024-11-02T00:00:00+01:00,C-3,2023-01-02,,
category,,,,C-3,,Electronics,20200115
category,,,,C-1,,Books,20191220

The Prague offset — one call with TZ_OFFSET

PASS 3 tags the ISO order date with its DST-aware Europe/Prague offset — CET (+01:00) in winter, CEST (+02:00) in summer. TZ_OFFSET reads the correct offset from the bundled IANA tz database, so daylight-saving time is handled with no hand-rolled calendar math:

IF(LEN([order_date]) = 0, '',                                          // (1)!
   [order_date] & 'T00:00:00' & TZ_OFFSET([order_date], 'Europe/Prague'))  // (2)!

1. Empty-guard — category rows (no order_date) and startup validation stay safe.
2. DST-aware ±HH:MM offset for the date — +01:00 in January, +02:00 in July.

Under the hood — deriving the offset by hand

Before the timezone builtins existed you'd compute the same offset from calendar primitives. EU clocks switch on the last Sunday of March / October (Directive 2000/84/EC); NTH_DOW(YEAR([order_date]), 3, 7, -1) returns the last Sunday of March directly (ISO weekday 7, occurrence -1 = last), and a DATEDIFF in-window test picks the summer or winter offset:

IF(DATEDIFF([order_date], NTH_DOW(YEAR([order_date]), 3, 7, -1)) >= 0
   AND DATEDIFF([order_date], NTH_DOW(YEAR([order_date]), 10, 7, -1)) < 0,
   '+02:00', '+01:00')

TZ_OFFSET([order_date], 'Europe/Prague') gives the identical result in one call. The dedicated timezone-functions example walks through all four TZ builtins (TO_UTC / TZ_OFFSET / IS_DST / TZ_CONVERT).

Pass 4 · join_category — hop 2, category → name

The last pass indexes the category rows and resolves category_name plus the created date (long YYYYMMDD → ISO). Because passes 2 and 3 did the format work, this one is pure lookup and passthrough — see Final result below.

Final result

Three orders, three source date formats, two join hops, and a timezone that flips with the season — one clean JSON dataset. An order arrives knowing only a product_id and a US date, and leaves knowing its category's name and its own Prague instant:

order 1001   product_id P-9   order_date 01/15/2024   (that is all it knew)
             ↓ hop 1 · category_id C-3      ↓ hop 2 · category_name Electronics
             order_ts 2024-01-15T00:00:00+01:00   category_created 2020-01-15

Watch order_ts across the three orders: +01:00 in January, +02:00 in July, back to +01:00 in November — one TZ_OFFSET call, no calendar arithmetic. The whole three-object result is in the 1-final.json (result) tab below (bxp writes one compact object per line).

Sample data

Run it with bxp-cli --config ./sample.json — the three inputs and the full commented template:

sample.json (config)orders.csvproducts.csvcategories.in.json1-final.json (result)

{
  // Teaching example — synthetic data. The CAPSTONE: a transitive (two-hop) JOIN
  // that one pass cannot do, chained across passes, while harmonising three
  // different date formats and bridging CSV <-> JSON.
  //
  // The graph: order -> product -> category -> name. An order knows only its
  // product_id; the product knows its category_id; the category name lives in a
  // third file. The second hop's key (category_id) does not exist on an order
  // until the first hop has run — so it cannot be done in a single pre_pass.
  //
  // bxp reads one file per pass and runs one pre_pass per pass, so the pipeline
  // is: combine -> join (hop 1) -> combine -> join (hop 2). Run it all with:
  //   bxp-cli --config ./sample.json
  //
  // Each source carries its own date format (the author knows which is which):
  //   orders.csv        order_date  US    MM/DD/YYYY
  //   products.csv      added_date  EU    DD.MM.YYYY
  //   categories.json   created     long  YYYYMMDD
  // All three are normalised to ISO. The order timestamp additionally gets a
  // DST-aware Europe/Prague offset (+01:00 winter / +02:00 summer) computed
  // purely from date builtins — see the big $order_ts expression in PASS 4.
  //
  // Glob safety: intermediate JSONs end ".in.json" so a re-run never re-reads
  // the final ".json"; combined files use the "1-<tpl>-combined.csvx" name.
  conversion_templates: {

    // PASS 1 (combine) — stack orders + products into one file with a `_type`
    // marker so the next pass's pre_pass can index across both. `_type` is
    // derived from which key the row has (orders have order_id, products don't).
    combine_op: {
      data_dir:          ".",
      file_pattern_in:   ".csv",
      file_pattern_out:  ".csvx",
      combined_output:   true,
      input_schema: {
        $_type:       "IF(LEN([order_id]) > 0, 'order', 'product')",
        $order_id:    "[order_id]",
        $product_id:  "[product_id]",
        $order_date:  "[order_date]",
        $category_id: "[category_id]",
        $added_date:  "[added_date]"
      },
      row_rules: [ { when: "1 = 1", rows: [ {} ] } ],
      output_schema: {
        _type: "$_type", order_id: "$order_id", product_id: "$product_id",
        order_date: "$order_date", category_id: "$category_id", added_date: "$added_date"
      }
    },

    // PASS 2 (hop 1) — index products, attach category_id (+ product added_date
    // converted EU->ISO) to each order. Convert order_date US->ISO here too.
    // Output as JSON so it can be combined with the categories JSON next.
    join_product: {
      data_dir:          ".",
      file_pattern_in:   "combine_op-combined.csvx",
      file_type_out:     "json",
      file_pattern_out:  "enriched.in.json",
      pre_pass: {
        when:   "[_type] = 'product'",
        key:    "[product_id]",
        values: {
          cat_id:     "[category_id]",
          padded_iso: "DATE_CONVERT([added_date], 'DD.MM.YYYY', 'YYYY-MM-DD')"
        }
      },
      input_schema: {
        $order_id:      "[order_id]",
        $product_id:    "[product_id]",
        $order_date:    "DATE_CONVERT([order_date], 'MM/DD/YYYY', 'YYYY-MM-DD')",
        $category_id:   "LOOKUP([product_id], 'cat_id')",
        $product_added: "LOOKUP([product_id], 'padded_iso')"
      },
      row_rules: [ { when: "[_type] = 'order'", rows: [ {} ] } ],
      output_schema: {
        order_id: "$order_id", product_id: "$product_id", order_date: "$order_date",
        category_id: "$category_id", product_added: "$product_added"
      }
    },

    // PASS 3 (combine) — stack the enriched orders (JSON) with the categories
    // (JSON) into one file. `_type` again derived from key presence.
    combine_oc: {
      data_dir:          ".",
      file_type_in:      "json",
      file_pattern_in:   ".in.json",
      file_type_out:     "csv",
      file_pattern_out:  ".csvx",
      combined_output:   true,
      input_schema: {
        $_type:         "IF(LEN([order_id]) > 0, 'order', 'category')",
        $order_id:      "[order_id]",
        $product_id:    "[product_id]",
        // Build the DST-aware Europe/Prague timestamp HERE, where order_date is
        // already ISO (converted in PASS 2) and present as a field — so the final
        // join pass stays free of date math. TZ_OFFSET resolves the correct
        // CET (+01:00) / CEST (+02:00) offset from the bundled IANA tz database,
        // DST and all. Empty-guarded so category rows (no order_date) and startup
        // validation stay safe. (Before the timezone builtins existed you'd
        // derive this by hand from NTH_DOW — see the basic timezone-functions
        // example; TZ_OFFSET gives the identical result in one call.)
        $order_ts:      "IF(LEN([order_date]) = 0, '', [order_date] & 'T00:00:00' & TZ_OFFSET([order_date], 'Europe/Prague'))",
        $category_id:   "[category_id]",
        $product_added: "[product_added]",
        $category_name: "[category_name]",
        $created:       "[created]"
      },
      row_rules: [ { when: "1 = 1", rows: [ {} ] } ],
      output_schema: {
        _type: "$_type", order_id: "$order_id", product_id: "$product_id",
        order_ts: "$order_ts", category_id: "$category_id",
        product_added: "$product_added", category_name: "$category_name", created: "$created"
      }
    },

    // PASS 4 (hop 2) — index categories, attach category_name and the created
    // date (long YYYYMMDD -> ISO). The order timestamp was already built in
    // PASS 3, so this final pass is pure lookup + passthrough — no date math.
    join_category: {
      data_dir:          ".",
      file_pattern_in:   "combine_oc-combined.csvx",
      file_type_out:     "json",
      file_pattern_out:  "final.json",
      pre_pass: {
        when:   "[_type] = 'category'",
        key:    "[category_id]",
        values: {
          cat_name:    "[category_name]",
          cat_created: "DATE_CONVERT([created], 'YYYYMMDD', 'YYYY-MM-DD')"
        }
      },
      input_schema: {
        $order_id:         "[order_id]",
        $product_id:       "[product_id]",
        $category_id:      "[category_id]",
        $category_name:    "LOOKUP([category_id], 'cat_name')",
        $product_added:    "[product_added]",
        $category_created: "LOOKUP([category_id], 'cat_created')",
        $order_ts:         "[order_ts]"
      },
      row_rules: [ { when: "[_type] = 'order'", rows: [ {} ] } ],
      output_schema: {
        order_id: "$order_id", product_id: "$product_id", category_id: "$category_id",
        category_name: "$category_name", order_ts: "$order_ts",
        product_added: "$product_added", category_created: "$category_created"
      }
    }
  }
}

order_id,product_id,order_date
1001,P-9,01/15/2024
1002,P-7,07/20/2024
1003,P-9,11/02/2024

product_id,category_id,added_date
P-9,C-3,02.01.2023
P-7,C-1,15.06.2023

[
  {"category_id":"C-3","category_name":"Electronics","created":"20200115"},
  {"category_id":"C-1","category_name":"Books","created":"20191220"}
]

[
{"order_id":"1001","product_id":"P-9","category_id":"C-3","category_name":"Electronics","order_ts":"2024-01-15T00:00:00+01:00","product_added":"2023-01-02","category_created":"2020-01-15"},
{"order_id":"1002","product_id":"P-7","category_id":"C-1","category_name":"Books","order_ts":"2024-07-20T00:00:00+02:00","product_added":"2023-06-15","category_created":"2019-12-20"},
{"order_id":"1003","product_id":"P-9","category_id":"C-3","category_name":"Electronics","order_ts":"2024-11-02T00:00:00+01:00","product_added":"2023-01-02","category_created":"2020-01-15"}
]