Pipeline Examples

This guide provides comprehensive, real-world pipeline examples in all supported formats: YAML, JSON, and HOCON.

Format Comparison

The same pipeline expressed in all three supported configuration formats:

YAMLJSONHOCON

name: customer-daily-load
tenancy: acme-corp
namespace: production
dataset: customers
executionMode: batch
platformProfile: databricks
schedule: "0 6 * * *"
stages:
  - name: acquire-customers
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT5M
    config:
      path: /data/input/customers.csv
      format: csv
  - name: transform-customers
    type: transformation
    capability: transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: "SELECT id, name, UPPER(city) AS city FROM {input} WHERE active = true"
  - name: write-output
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite
policies: []
extensions: []

{
  "name": "customer-daily-load",
  "tenancy": "acme-corp",
  "namespace": "production",
  "dataset": "customers",
  "executionMode": "batch",
  "platformProfile": "databricks",
  "schedule": "0 6 * * *",
  "stages": [
    {
      "name": "acquire-customers",
      "type": "source",
      "capability": "source-acquisition",
      "provider": "spark-file-source",
      "timeout": "PT5M",
      "config": {
        "path": "/data/input/customers.csv",
        "format": "csv"
      }
    },
    {
      "name": "transform-customers",
      "type": "transformation",
      "capability": "transformation",
      "timeout": "PT10M",
      "transformation": {
        "sql": {
          "style": "sql",
          "reference": "SELECT id, name, UPPER(city) AS city FROM {input} WHERE active = true"
        }
      }
    },
    {
      "name": "write-output",
      "type": "publish",
      "capability": "storage",
      "provider": "spark-file-storage",
      "timeout": "PT5M",
      "config": {
        "format": "parquet",
        "mode": "overwrite"
      }
    }
  ],
  "policies": [],
  "extensions": []
}

name = "customer-daily-load"
tenancy = "acme-corp"
namespace = "production"
dataset = "customers"
executionMode = "batch"
platformProfile = "databricks"
schedule = "0 6 * * *"
stages = [
  {
    name = "acquire-customers"
    type = "source"
    capability = "source-acquisition"
    provider = "spark-file-source"
    timeout = "PT5M"
    config {
      path = "/data/input/customers.csv"
      format = "csv"
    }
  },
  {
    name = "transform-customers"
    type = "transformation"
    capability = "transformation"
    timeout = "PT10M"
    transformation {
      sql {
        style = "sql"
        reference = "SELECT id, name, UPPER(city) AS city FROM {input} WHERE active = true"
      }
    }
  },
  {
    name = "write-output"
    type = "publish"
    capability = "storage"
    provider = "spark-file-storage"
    timeout = "PT5M"
    config {
      format = "parquet"
      mode = "overwrite"
    }
  }
]
policies = []
extensions = []

Choosing a format

YAML is the recommended default — concise and human-readable. JSON works well for programmatic generation. HOCON is ideal for teams familiar with Typesafe/Lightbend ecosystems and offers features like variable substitution and includes.

---

File ETL Examples

1. CSV to Parquet

Basic ETL: read CSV files, filter rows, select columns, write as Parquet.

name: csv-to-parquet
tenancy: data-team
namespace: production
dataset: sales-records
executionMode: batch
schedule: "0 2 * * *"

stages:
  - name: read-csv-files
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/raw/sales/
      format: csv
      header: true
      inferSchema: true
      delimiter: ","

  - name: filter-and-project
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            order_id, customer_id, product_name,
            quantity, unit_price, 
            quantity * unit_price AS total_amount,
            order_date
          FROM {input}
          WHERE quantity > 0 AND unit_price > 0

  - name: write-parquet
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite
      partitionBy: order_date

policies: []
extensions: []

Key points:

• inferSchema: true lets Spark detect column types from CSV data
• The transformation filters invalid rows and computes a derived column
• Output is partitioned by order_date for efficient downstream queries

---

2. JSON Events to Delta Lake

Process JSON event files with data quality validation before writing to Delta Lake.

name: events-to-delta
tenancy: analytics-team
namespace: production
dataset: user-events
executionMode: batch
platformProfile: databricks
schedule: "*/15 * * * *"

stages:
  - name: ingest-json-events
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/landing/events/
      format: json
      multiLine: true

  - name: validate-events
    type: quality
    capability: data-quality
    timeout: PT3M

  - name: enrich-events
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            event_id, user_id, event_type,
            event_timestamp,
            CAST(event_timestamp AS DATE) AS event_date,
            properties,
            current_timestamp() AS processed_at
          FROM {input}
          WHERE event_id IS NOT NULL

  - name: write-delta-lake
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: append
      partitionBy: event_date

policies: []
extensions: []

Key points:

• multiLine: true handles JSON files with pretty-printed objects
• Quality validation stage catches malformed events before processing
• Delta Lake append mode supports incremental writes with ACID guarantees

---

3. Multi-Source Merge

Read from multiple directories, union the data, and aggregate.

name: multi-source-merge
tenancy: data-platform
namespace: production
dataset: unified-transactions
executionMode: batch
schedule: "0 4 * * *"

stages:
  - name: acquire-online-sales
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/raw/online-sales/
      format: parquet

  - name: acquire-store-sales
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/raw/store-sales/
      format: parquet

  - name: union-and-aggregate
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT15M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            region,
            product_category,
            SUM(amount) AS total_revenue,
            COUNT(*) AS transaction_count,
            AVG(amount) AS avg_transaction_value
          FROM {input}
          GROUP BY region, product_category

  - name: write-summary
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite
      partitionBy: region

policies: []
extensions: []

Key points:

• Multiple source stages feed into a single transformation
• The aggregation groups by region and product category
• Output is partitioned by region for efficient regional queries

---

4. Incremental File Load

File-based incremental processing with checkpoint tracking.

name: incremental-file-load
tenancy: data-team
namespace: production
dataset: order-updates
executionMode: batch
schedule: "0 * * * *"

stages:
  - name: acquire-new-files
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/landing/orders/
      format: parquet
      mode: incremental

  - name: deduplicate
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT * FROM (
            SELECT *,
              ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
            FROM {input}
          ) WHERE rn = 1

  - name: append-to-lake
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: append

policies: []
extensions: []

Key points:

• mode: incremental reads only files modified since the last checkpoint
• Deduplication ensures only the latest version of each order is kept
• Hourly schedule with append mode builds up the Delta Lake incrementally

---

Database ETL Examples

5. Full Table Extract

Snapshot a database table to Parquet for analytics.

name: full-table-extract
tenancy: analytics
namespace: production
dataset: product-catalog
executionMode: batch
schedule: "0 1 * * *"

stages:
  - name: extract-products
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT15M
    config:
      url: jdbc:postgresql://prod-db:5432/catalog
      table: products
      driver: org.postgresql.Driver
      user: reader
      password: ${DB_PASSWORD}
      fetchSize: 5000

  - name: write-snapshot
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite

policies: []
extensions: []

Key points:

• Full table snapshot runs nightly for analytics consumption
• fetchSize: 5000 optimises JDBC read performance
• Credentials use environment variable substitution (${DB_PASSWORD})

---

6. Incremental Database Sync

Watermark-based incremental load from a source database.

name: incremental-db-sync
tenancy: data-platform
namespace: production
dataset: customer-orders
executionMode: batch
schedule: "*/30 * * * *"

stages:
  - name: acquire-new-orders
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT10M
    config:
      url: jdbc:postgresql://orders-db:5432/orders
      table: customer_orders
      driver: org.postgresql.Driver
      user: etl_reader
      password: ${ORDERS_DB_PASSWORD}
      watermarkColumn: updated_at
      watermarkValue: "2024-01-15T00:00:00"

  - name: standardise-timestamps
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT *,
            CAST(updated_at AS TIMESTAMP) AS sync_timestamp,
            current_timestamp() AS ingested_at
          FROM {input}

  - name: append-to-lake
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: append

policies: []
extensions: []

Key points:

• watermarkColumn and watermarkValue enable incremental reads (only rows where updated_at > watermarkValue)
• The watermark value is typically managed by the Echo service and updated after each successful run
• Runs every 30 minutes to keep the data lake near real-time

---

7. Database to Database

Extract from source database, transform, and load into a target database.

name: db-to-db-sync
tenancy: data-platform
namespace: production
dataset: customer-summary
executionMode: batch
schedule: "0 3 * * *"

stages:
  - name: extract-orders
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT15M
    config:
      url: jdbc:postgresql://source-db:5432/transactions
      table: orders
      driver: org.postgresql.Driver
      user: reader
      password: ${SOURCE_DB_PASSWORD}
      fetchSize: 10000

  - name: aggregate-by-customer
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            customer_id,
            COUNT(*) AS total_orders,
            SUM(amount) AS lifetime_value,
            MAX(order_date) AS last_order_date,
            MIN(order_date) AS first_order_date
          FROM {input}
          GROUP BY customer_id

  - name: load-to-analytics-db
    type: publish
    capability: storage
    provider: spark-jdbc-storage
    timeout: PT10M
    config:
      url: jdbc:postgresql://analytics-db:5432/warehouse
      table: customer_summary
      driver: org.postgresql.Driver
      mode: overwrite
      batchSize: 5000

policies: []
extensions: []

Key points:

• Source and target are different databases — common in data warehouse patterns
• Aggregation reduces millions of order rows to one row per customer
• batchSize: 5000 optimises bulk insert performance

---

8. Partitioned Large Table Read

Parallel JDBC reads for large tables using partition columns.

name: large-table-parallel-read
tenancy: data-platform
namespace: production
dataset: transaction-history
executionMode: batch
schedule: "0 0 * * 0"

stages:
  - name: parallel-extract
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT60M
    config:
      url: jdbc:postgresql://warehouse:5432/history
      table: transactions
      driver: org.postgresql.Driver
      user: bulk_reader
      password: ${WAREHOUSE_PASSWORD}
      partitionColumn: id
      lowerBound: 1
      upperBound: 50000000
      numPartitions: 32
      fetchSize: 10000

  - name: write-partitioned
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT15M
    config:
      format: parquet
      mode: overwrite
      partitionBy: transaction_year,transaction_month

policies: []
extensions: []

Key points:

• numPartitions: 32 creates 32 parallel JDBC connections for fast extraction
• partitionColumn, lowerBound, upperBound define how Spark splits the reads
• Weekly schedule for full historical snapshots
• Output partitioned by year/month for time-based query patterns

---

Transformation Patterns

9. Multi-Stage Pipeline

Source → Clean → Enrich → Aggregate → Publish — a classic data engineering pattern.

name: multi-stage-etl
tenancy: data-platform
namespace: production
dataset: revenue-report
executionMode: batch
platformProfile: databricks
schedule: "0 5 * * *"

stages:
  - name: acquire-transactions
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/lake/transactions/
      format: delta

  - name: clean-data
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT * FROM {input}
          WHERE amount IS NOT NULL
            AND amount > 0
            AND customer_id IS NOT NULL
            AND transaction_date >= '2024-01-01'

  - name: enrich-with-customer
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            t.*,
            c.segment AS customer_segment,
            c.region AS customer_region
          FROM {input} t
          JOIN customer_dim c ON t.customer_id = c.id

  - name: aggregate-revenue
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            customer_region,
            customer_segment,
            CAST(transaction_date AS DATE) AS report_date,
            SUM(amount) AS total_revenue,
            COUNT(DISTINCT customer_id) AS unique_customers,
            COUNT(*) AS transaction_count
          FROM {input}
          GROUP BY customer_region, customer_segment, CAST(transaction_date AS DATE)

  - name: publish-report
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: overwrite
      partitionBy: report_date

policies: []
extensions: []

Key points:

• Each stage has a single responsibility: clean, enrich, aggregate
• Intermediate results flow automatically between stages
• The enrichment join adds dimensional attributes for reporting
• Final output is a date-partitioned Delta table for dashboards

---

10. Data Quality Pipeline

Source → Validate → Route (quarantine bad records / pass good records) → Write.

name: quality-gated-pipeline
tenancy: data-governance
namespace: production
dataset: validated-customers
executionMode: batch
schedule: "0 3 * * *"

stages:
  - name: acquire-raw-customers
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT5M
    config:
      path: /data/landing/customers/
      format: csv
      header: true
      inferSchema: true

  - name: validate-data-quality
    type: quality
    capability: data-quality
    timeout: PT5M

  - name: filter-valid-records
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT * FROM {input}
          WHERE email IS NOT NULL
            AND email LIKE '%@%.%'
            AND LENGTH(name) > 0
            AND age BETWEEN 0 AND 150

  - name: write-validated
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite

policies: []
extensions: []

Key points:

• Quality stage runs framework-level validation rules
• Transformation stage applies business-specific validation logic
• Invalid records are filtered out (could also be routed to a quarantine table)
• Only clean data reaches the final output

---

11. Deduplication Pipeline

Remove duplicate records based on a business key, keeping the most recent version.

name: deduplication-pipeline
tenancy: data-platform
namespace: production
dataset: unique-events
executionMode: batch
schedule: "0 2 * * *"

stages:
  - name: acquire-events
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/raw/events/
      format: parquet
      mergeSchema: true

  - name: deduplicate-by-key
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT * FROM (
            SELECT *,
              ROW_NUMBER() OVER (
                PARTITION BY event_id 
                ORDER BY event_timestamp DESC
              ) AS row_num
            FROM {input}
          ) ranked
          WHERE row_num = 1

  - name: write-deduplicated
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: overwrite

policies: []
extensions: []

Key points:

• ROW_NUMBER() window function keeps only the latest record per event_id
• mergeSchema: true handles schema evolution across Parquet files
• Common pattern for event sourcing systems where duplicates are expected

---

12. Pivot/Unpivot for Reporting

Reshape data from long format to wide format for reporting dashboards.

name: pivot-for-reporting
tenancy: analytics
namespace: production
dataset: monthly-metrics-wide
executionMode: batch
schedule: "0 6 1 * *"

stages:
  - name: acquire-metrics
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT5M
    config:
      path: /data/lake/monthly-metrics/
      format: parquet

  - name: pivot-metrics
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            department,
            SUM(CASE WHEN metric_name = 'revenue' THEN metric_value ELSE 0 END) AS revenue,
            SUM(CASE WHEN metric_name = 'cost' THEN metric_value ELSE 0 END) AS cost,
            SUM(CASE WHEN metric_name = 'headcount' THEN metric_value ELSE 0 END) AS headcount,
            SUM(CASE WHEN metric_name = 'revenue' THEN metric_value ELSE 0 END) -
              SUM(CASE WHEN metric_name = 'cost' THEN metric_value ELSE 0 END) AS profit
          FROM {input}
          WHERE report_month = '2024-01'
          GROUP BY department

  - name: write-wide-report
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: parquet
      mode: overwrite

policies: []
extensions: []

Key points:

• Conditional aggregation (CASE WHEN) pivots rows into columns
• Derived column (profit) computed inline
• Monthly schedule runs on the 1st of each month
• Wide format is optimised for BI tool consumption

---

Streaming Examples (Flink)

13. File Monitoring

Watch a directory for new files and process them continuously.

name: file-monitor-stream
tenancy: data-platform
namespace: production
dataset: realtime-orders
executionMode: streaming
platformProfile: aws-native

stages:
  - name: monitor-landing-zone
    type: source
    capability: source-acquisition
    provider: flink-file-source
    timeout: PT0S
    config:
      path: /data/landing/orders/
      format: json
      monitorInterval: PT30S

  - name: parse-and-validate
    type: transformation
    capability: transformation
    provider: flink-transformation
    timeout: PT0S
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            order_id, customer_id, amount, order_timestamp,
            CASE WHEN amount > 0 THEN 'valid' ELSE 'invalid' END AS status
          FROM {input}
          WHERE order_id IS NOT NULL

  - name: write-to-lake
    type: publish
    capability: storage
    provider: flink-file-sink
    timeout: PT0S
    config:
      format: parquet
      mode: append
      checkpointInterval: PT1M

policies: []
extensions: []

Key points:

• executionMode: streaming enables continuous processing
• monitorInterval: PT30S checks for new files every 30 seconds
• timeout: PT0S means stages run indefinitely (streaming semantics)
• Checkpoint interval ensures exactly-once delivery guarantees

---

14. Windowed Aggregation

5-minute tumbling window aggregation for real-time metrics.

name: windowed-metrics
tenancy: analytics
namespace: production
dataset: realtime-metrics
executionMode: streaming
platformProfile: aws-native

stages:
  - name: consume-events
    type: source
    capability: source-acquisition
    provider: flink-kafka-source
    timeout: PT0S
    config:
      bootstrapServers: kafka-cluster:9092
      topic: user-events
      groupId: metrics-aggregator
      startingOffsets: latest

  - name: window-aggregate
    type: transformation
    capability: transformation
    provider: flink-transformation
    timeout: PT0S
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            event_type,
            TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
            TUMBLE_END(event_time, INTERVAL '5' MINUTE) AS window_end,
            COUNT(*) AS event_count,
            COUNT(DISTINCT user_id) AS unique_users
          FROM {input}
          GROUP BY event_type, TUMBLE(event_time, INTERVAL '5' MINUTE)

  - name: emit-metrics
    type: publish
    capability: storage
    provider: flink-kafka-sink
    timeout: PT0S
    config:
      bootstrapServers: kafka-cluster:9092
      topic: aggregated-metrics

policies: []
extensions: []

Key points:

• Kafka source with startingOffsets: latest for real-time processing
• TUMBLE window function creates non-overlapping 5-minute windows
• Output goes back to Kafka for downstream consumers (dashboards, alerts)
• Flink handles watermarks and late data automatically

---

Advanced Patterns

15. Dual-Mode Transformation

Use both Spark DataFrame API and SQL for platform portability.

name: dual-mode-pipeline
tenancy: data-platform
namespace: production
dataset: enriched-orders
executionMode: batch
platformProfile: databricks

stages:
  - name: acquire-orders
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT10M
    config:
      url: jdbc:postgresql://orders-db:5432/orders
      table: customer_orders
      driver: org.postgresql.Driver
      user: reader
      password: ${DB_PASSWORD}

  - name: complex-enrichment
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT20M
    retryPolicy:
      maxAttempts: 2
      backoff: PT30S
      strategy: exponential
    transformation:
      spark:
        style: dataframe
        reference: com.acme.transforms.OrderEnrichment
        config:
          lookupTable: customer_segments
          geocodingEnabled: true
      sql:
        style: sql
        reference: transforms/order_enrichment.sql

  - name: write-enriched
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    config:
      format: delta
      mode: overwrite
      partitionBy: order_date

policies: []
extensions: []

Key points:

• spark mode uses a custom DataFrame class for complex logic (geocoding, ML scoring)
• sql mode provides a portable fallback for platforms without Spark
• The runtime selects the appropriate mode based on the platform profile
• Retry policy handles transient failures in the enrichment stage

---

16. Pipeline with Retry Policy

Resilient pipeline with exponential backoff for unreliable sources.

name: resilient-api-ingest
tenancy: integrations
namespace: production
dataset: partner-data
executionMode: batch
schedule: "0 */4 * * *"

stages:
  - name: fetch-partner-data
    type: source
    capability: source-acquisition
    provider: spark-jdbc-source
    timeout: PT15M
    retryPolicy:
      maxAttempts: 5
      backoff: PT10S
      strategy: exponential
    config:
      url: jdbc:postgresql://partner-db:5432/shared
      table: partner_feed
      driver: org.postgresql.Driver
      user: integration_user
      password: ${PARTNER_DB_PASSWORD}
      connectionTimeout: 60

  - name: normalise-schema
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT10M
    retryPolicy:
      maxAttempts: 3
      backoff: PT5S
      strategy: exponential
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            COALESCE(partner_id, 'UNKNOWN') AS partner_id,
            TRIM(UPPER(product_code)) AS product_code,
            CAST(quantity AS INT) AS quantity,
            CAST(price AS DECIMAL(10,2)) AS price,
            COALESCE(currency, 'USD') AS currency,
            current_timestamp() AS ingested_at
          FROM {input}

  - name: write-normalised
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT5M
    retryPolicy:
      maxAttempts: 3
      backoff: PT5S
      strategy: exponential
    config:
      format: parquet
      mode: append

policies: []
extensions: []

Key points:

• Every stage has a retry policy — essential for unreliable external sources
• Source stage has 5 attempts with exponential backoff (10s, 20s, 40s, 80s, 160s)
• connectionTimeout: 60 gives the partner database time to respond
• Schema normalisation handles nulls and type coercion defensively

---

17. Partitioned Output

Write with date and region partitioning for efficient query patterns.

name: partitioned-output-pipeline
tenancy: data-platform
namespace: production
dataset: regional-sales
executionMode: batch
platformProfile: databricks
schedule: "0 4 * * *"

stages:
  - name: acquire-sales
    type: source
    capability: source-acquisition
    provider: spark-file-source
    timeout: PT10M
    config:
      path: /data/raw/daily-sales/
      format: parquet

  - name: add-partition-columns
    type: transformation
    capability: transformation
    provider: spark-transformation
    timeout: PT5M
    transformation:
      sql:
        style: sql
        reference: |
          SELECT 
            *,
            CAST(sale_date AS DATE) AS partition_date,
            COALESCE(region, 'UNKNOWN') AS partition_region,
            YEAR(sale_date) AS sale_year,
            MONTH(sale_date) AS sale_month
          FROM {input}

  - name: write-partitioned
    type: publish
    capability: storage
    provider: spark-file-storage
    timeout: PT10M
    config:
      format: delta
      mode: overwrite
      partitionBy: sale_year,sale_month,partition_region

policies: []
extensions: []

Key points:

• Partition columns are computed in the transformation stage
• Three-level partitioning: year/month/region enables efficient pruning
• COALESCE ensures no null partition values (which cause issues in Hive-style partitioning)
• Delta format with overwrite replaces the full dataset daily

---

HOCON Examples

For teams using the Typesafe/Lightbend ecosystem, here are select examples in HOCON format.

Multi-Stage Pipeline in HOCON

name = "multi-stage-etl"
tenancy = "data-platform"
namespace = "production"
dataset = "revenue-report"
executionMode = "batch"
platformProfile = "databricks"
schedule = "0 5 * * *"

stages = [
  {
    name = "acquire-transactions"
    type = "source"
    capability = "source-acquisition"
    provider = "spark-file-source"
    timeout = "PT10M"
    config {
      path = "/data/lake/transactions/"
      format = "delta"
    }
  },
  {
    name = "clean-data"
    type = "transformation"
    capability = "transformation"
    provider = "spark-transformation"
    timeout = "PT5M"
    transformation {
      sql {
        style = "sql"
        reference = """
          SELECT * FROM {input}
          WHERE amount IS NOT NULL AND amount > 0
        """
      }
    }
  },
  {
    name = "publish-report"
    type = "publish"
    capability = "storage"
    provider = "spark-file-storage"
    timeout = "PT5M"
    config {
      format = "delta"
      mode = "overwrite"
    }
  }
]

policies = []
extensions = []

Resilient Pipeline in HOCON

name = "resilient-ingest"
tenancy = "integrations"
namespace = "production"
dataset = "partner-data"
executionMode = "batch"
schedule = "0 */4 * * *"

stages = [
  {
    name = "fetch-partner-data"
    type = "source"
    capability = "source-acquisition"
    provider = "spark-jdbc-source"
    timeout = "PT15M"
    retryPolicy {
      maxAttempts = 5
      backoff = "PT10S"
      strategy = "exponential"
    }
    config {
      url = "jdbc:postgresql://partner-db:5432/shared"
      table = "partner_feed"
      driver = "org.postgresql.Driver"
      user = "integration_user"
      password = ${PARTNER_DB_PASSWORD}
      connectionTimeout = 60
    }
  },
  {
    name = "write-output"
    type = "publish"
    capability = "storage"
    provider = "spark-file-storage"
    timeout = "PT5M"
    config {
      format = "parquet"
      mode = "append"
    }
  }
]

policies = []
extensions = []

HOCON variable substitution

HOCON natively supports ${VARIABLE} syntax for environment variable references. This is resolved at parse time by the Typesafe Config library.

---

Next Steps

• Writing Pipelines — Pipeline structure reference
• Stage Types — All available stage types
• Apache Spark Provider Guide — Detailed Spark configuration
• Apache Flink Provider Guide — Streaming with Flink
• Dual-Mode Transformations — Platform-portable transforms