Stop a Bad Monthly Load with SLA Assertions on NYC Taxi Data in DuckDB

id: nyc-taxi-monthly-load-sla-circuit-breaker
namespace: company.team

inputs:
  - id: month
    type: STRING
    displayName: Month to load
    description: A month of NYC TLC green taxi trips, as `yyyy-MM`. Leave it empty
      for the month two months back, the latest TLC has usually published.
    required: false
    validator: ^20[0-9]{2}-(0[1-9]|1[0-2])$

  - id: min_trips
    type: INT
    displayName: Minimum trips
    description: Fewer trips than this means a truncated or empty file.
    defaults: 10000

  - id: max_change_pct
    type: FLOAT
    displayName: Maximum change from the previous month (%)
    description: Checked only when the previous month was loaded by this flow.
    defaults: 50

  - id: max_null_passenger_pct
    type: FLOAT
    displayName: Maximum share of trips without a passenger count (%)
    defaults: 25

  - id: max_out_of_month_pct
    type: FLOAT
    displayName: Maximum share of trips picked up outside the month (%)
    defaults: 1

variables:
  month: "{{ inputs.month ?? ((trigger.date ?? now()) | dateAdd(-2, 'MONTHS') |
    date('yyyy-MM')) }}"
  previous_month: "{{ (render(vars.month) ~ '-01') | dateAdd(-1, 'MONTHS') |
    date('yyyy-MM') }}"
  published_path: "nyc_taxi/green/{{ render(vars.month) }}.parquet"

# Each rule is checked every time the execution changes, from the first task on. Until `profile` has run its
# output doesn't exist, so `??` gives a value that passes; without it the rule can't be evaluated and Kestra
# treats it as violated, failing every run before the download.
sla:
  - id: enough_trips
    type: EXECUTION_ASSERTION
    behavior: FAIL
    assert: "{{ (outputs.profile.row.trips ?? inputs.min_trips) >= inputs.min_trips }}"
    labels:
      sla_breach: enough_trips

  - id: stable_volume
    type: EXECUTION_ASSERTION
    behavior: FAIL
    assert: "{{ ((outputs.profile.row.change_pct ?? 0) | abs) <=
      inputs.max_change_pct }}"
    labels:
      sla_breach: stable_volume

  - id: passenger_count_filled
    type: EXECUTION_ASSERTION
    behavior: FAIL
    assert: "{{ (outputs.profile.row.null_passenger_pct ?? 0) <=
      inputs.max_null_passenger_pct }}"
    labels:
      sla_breach: passenger_count_filled

  - id: trips_in_month
    type: EXECUTION_ASSERTION
    behavior: FAIL
    assert: "{{ (outputs.profile.row.out_of_month_pct ?? 0) <=
      inputs.max_out_of_month_pct }}"
    labels:
      sla_breach: trips_in_month

  - id: load_duration
    type: MAX_DURATION
    behavior: FAIL
    duration: PT15M
    labels:
      sla_breach: load_duration

tasks:
  - id: fetch
    type: io.kestra.plugin.core.flow.Parallel
    description: Reads last month's baseline and downloads this month side by side.
    tasks:
      - id: previous
        type: io.kestra.plugin.core.kv.Get
        description: Trip count of the previous month, saved by the run that published it.
        key: "nyc_taxi_green_stats_{{ render(vars.previous_month) }}"
        errorOnMissing: false

      - id: download
        type: io.kestra.plugin.core.http.Download
        uri: "https://d37ci6vzurychx.cloudfront.net/trip-data/green_tripdata_{{
          render(vars.month) }}.parquet"

  - id: profile
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: The numbers the SLA rules read. Nothing is published yet.
    inputFiles:
      trips.parquet: "{{ outputs.download.uri }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT count(*) AS trips,
             {{ outputs.previous.value.trips ?? 'NULL' }} AS previous_trips,
             round(100.0 * (count(*) - {{ outputs.previous.value.trips ?? 'NULL' }}) / {{ outputs.previous.value.trips ?? 'NULL' }}, 1) AS change_pct,
             round(100.0 * count(*) FILTER (WHERE passenger_count IS NULL) / count(*), 2) AS null_passenger_pct,
             round(100.0 * count(*) FILTER (WHERE strftime(lpep_pickup_datetime, '%Y-%m') <> '{{ render(vars.month) }}') / count(*), 3) AS out_of_month_pct
      FROM '{{ workingDir }}/trips.parquet'

  - id: curate
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Keeps the trips picked up in the month, with a distance and a
      non-negative fare.
    inputFiles:
      trips.parquet: "{{ outputs.download.uri }}"
    outputFiles:
      - curated.parquet
    sql: |
      COPY (
        SELECT * FROM '{{ workingDir }}/trips.parquet'
        WHERE strftime(lpep_pickup_datetime, '%Y-%m') = '{{ render(vars.month) }}'
          AND trip_distance > 0
          AND fare_amount >= 0
        ORDER BY lpep_pickup_datetime
      ) TO '{{ outputFiles["curated.parquet"] }}' (FORMAT parquet)

  - id: publish
    type: io.kestra.plugin.core.namespace.UploadFiles
    description: The published file other flows read, in this namespace's files.
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.published_path) }}": "{{ outputs.curate.outputFiles['curated.parquet'] }}"

  - id: remember
    type: io.kestra.plugin.core.kv.Set
    description: The baseline the next month is compared with.
    key: "nyc_taxi_green_stats_{{ render(vars.month) }}"
    kvType: JSON
    value: "{{ outputs.profile.row | toJson }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Published {{ render(vars.month) }} to {{ render(vars.published_path) }}:
      {{ outputs.profile.row.trips }} trips{{ outputs.profile.row.change_pct is
      not null ? ' (' ~ outputs.profile.row.change_pct ~ '% from ' ~
      render(vars.previous_month) ~ ')' : '' }}, {{
      outputs.profile.row.null_passenger_pct }}% without a passenger count, {{
      outputs.profile.row.out_of_month_pct }}% picked up outside the month.

outputs:
  - id: published_path
    type: STRING
    required: false
    value: "{{ render(vars.published_path) }}"

triggers:
  - id: monthly
    type: io.kestra.plugin.core.trigger.Schedule
    description: On the 25th, load the month two months back, which TLC has
      published by then.
    cron: "0 6 25 * *"