Monthly NYC Taxi Metrics with DuckDB as a Backfill Partition Job

id: backfill-example-nyc-taxi-monthly
namespace: company.team
description: |
  Example partition job for the safe-backfill-controller blueprint. It computes
  trip metrics for one month of NYC yellow taxi data straight from the public
  Parquet file with DuckDB, and stores the result under a key for that month, so
  re-running a month replaces its result instead of duplicating it.

inputs:
  - id: partition_start
    type: DATE
    displayName: Partition start
    description: First day of the month to process (inclusive).
    defaults: "2024-01-01"
  - id: partition_end
    type: DATE
    displayName: Partition end
    description: First day of the next month (exclusive).
    defaults: "2024-02-01"

triggers:
  - id: monthly_refresh
    type: io.kestra.plugin.core.trigger.Schedule
    description: Process the previous month on the 15th, once NYC TLC has published
      it. Backfills of older months go through the controller instead.
    cron: "0 6 15 * *"
    disabled: true
    inputs:
      partition_start: "{{ trigger.date | dateAdd(-1, 'MONTHS') | date('yyyy-MM-01') }}"
      partition_end: "{{ trigger.date | date('yyyy-MM-01') }}"

tasks:
  - id: monthly_metrics
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Read only the needed columns of the month's Parquet file over HTTP
      (DuckDB loads its httpfs extension automatically) and aggregate them.
      Brief network errors are retried. A month that isn't published yet still
      fails here, which is what the controller retries later.
    retry:
      type: exponential
      interval: PT10S
      maxInterval: PT1M
      maxAttempts: 3
    sql: |
      SELECT
        '{{ inputs.partition_start | date('yyyy-MM') }}' AS month,
        COUNT(*) AS trips,
        ROUND(AVG(trip_distance), 2) AS avg_distance_miles,
        ROUND(AVG(fare_amount), 2) AS avg_fare_usd,
        CAST(ROUND(SUM(total_amount)) AS BIGINT) AS revenue_usd
      FROM read_parquet('https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{{ inputs.partition_start | date('yyyy-MM') }}.parquet')
      WHERE tpep_pickup_datetime >= TIMESTAMP '{{ inputs.partition_start }}'
        AND tpep_pickup_datetime < TIMESTAMP '{{ inputs.partition_end }}'
    fetchType: FETCH_ONE

  - id: quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Refuse to store an empty month, so a partition with no data is
      never checkpointed as done.
    condition: "{{ outputs.monthly_metrics.row.trips == 0 }}"
    then:
      - id: empty_month
        type: io.kestra.plugin.core.execution.Fail
        description: Fail the partition so the backfill controller retries it later.
        errorMessage: "No trips found between {{ inputs.partition_start }} and {{
          inputs.partition_end }}."
    else:
      - id: store_result
        type: io.kestra.plugin.core.kv.Set
        description: Upsert the month's metrics under one key per month, so the job is
          idempotent and safe to re-run.
        key: "nyc_taxi_metrics_{{ inputs.partition_start | date('yyyy_MM') }}"
        kvType: JSON
        value: "{{ outputs.monthly_metrics.row | toJson }}"

      - id: log_result
        type: io.kestra.plugin.core.log.Log
        description: Record the month's headline numbers in the execution logs.
        message: "{{ outputs.monthly_metrics.row.month }}: {{
          outputs.monthly_metrics.row.trips }} trips, ${{
          outputs.monthly_metrics.row.revenue_usd }} revenue."

errors:
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    description: Explain the failure in this run's logs. No alert is sent from here,
      because the backfill controller already reports failed partitions.
    level: ERROR
    message: |
      Partition {{ inputs.partition_start ?? '' }} to {{ inputs.partition_end ?? '' }} failed in task {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the logs above' }}
      Nothing was stored for this month. Re-running the backfill with the same backfill ID retries it.

outputs:
  - id: metrics
    type: JSON
    displayName: Monthly metrics
    value: "{{ outputs.monthly_metrics.row }}"