Zero-Warehouse Lakehouse Curation and Parquet Export with DuckDB

id: duckdb-lakehouse-parquet-curation
namespace: company.team
description: |
  Automated zero-infrastructure Lakehouse curation pipeline using DuckDB. Ingests raw batch
  data, validates quality constraints, transforms records with analytical window functions,
  stores compressed columnar Parquet, and posts KPI digests to Slack.

concurrency:
  limit: 1

inputs:
  - id: data_url
    type: STRING
    displayName: Raw Data Source URL
    description: Public or pre-signed URL to the raw CSV dataset.
    defaults: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv"

  - id: min_expected_rows
    type: INT
    displayName: Minimum Expected Rows
    description: Quality check threshold to prevent processing truncated or empty feeds.
    defaults: 10

  - id: slack_channel
    type: STRING
    displayName: Alert Channel
    description: Slack channel destination for daily KPI scorecard.
    defaults: "#data-engineering"

triggers:
  - id: daily_lakehouse_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily curation run at 06:00 UTC. Can also be triggered manually
      with inputs.
    cron: "0 6 * * *"

tasks:
  - id: fetch_raw_batch
    type: io.kestra.plugin.core.http.Download
    description: Ingest raw daily batch file into Kestra managed storage.
    uri: "{{ inputs.data_url }}"

  - id: validate_and_profile
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Profile incoming data with in-memory DuckDB to enforce data quality
      constraints.
    sql: |
      INSTALL httpfs;
      LOAD httpfs;

      CREATE OR REPLACE TEMP TABLE raw_orders AS
      SELECT * FROM read_csv_auto('{{ outputs.fetch_raw_batch.uri }}', header=True);

      SELECT
        COUNT(*) AS total_rows,
        COUNT(CASE WHEN order_id IS NULL THEN 1 END) AS null_order_ids,
        COUNT(CASE WHEN total <= 0 THEN 1 END) AS invalid_totals,
        ROUND(COALESCE(SUM(total), 0), 2) AS gross_revenue,
        ROUND(COALESCE(AVG(total), 0), 2) AS avg_order_value,
        COUNT(DISTINCT customer_id) AS unique_customers
      FROM raw_orders;
    fetchType: FETCH

  - id: quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluate profiling metrics to ensure 100% data integrity before curation.
    condition: "{{ outputs.validate_and_profile.rows[0].total_rows >=
      inputs.min_expected_rows and
      outputs.validate_and_profile.rows[0].null_order_ids == 0 and
      outputs.validate_and_profile.rows[0].invalid_totals == 0 }}"
    then:
      - id: curate_and_export_parquet
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: Run window analytics, enrich schema, and persist columnar Parquet
          to storage.
        sql: |
          INSTALL parquet;
          LOAD parquet;

          CREATE OR REPLACE TEMP TABLE raw_orders AS
          SELECT * FROM read_csv_auto('{{ outputs.fetch_raw_batch.uri }}', header=True);

          SELECT
            order_id,
            customer_id,
            total,
            quantity,
            CURRENT_TIMESTAMP AS curated_at,
            AVG(total) OVER (PARTITION BY customer_id) AS customer_historical_avg,
            DENSE_RANK() OVER (ORDER BY total DESC) AS revenue_rank
          FROM raw_orders
          ORDER BY total DESC;
        fetchType: STORE

      - id: notify_slack_kpi
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Broadcast curated Lakehouse KPIs and quality scorecard to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *Lakehouse Daily Curation Complete*
          • *Status*: Passed Data Quality Gates
          • *Processed Rows*: `{{ outputs.validate_and_profile.rows[0].total_rows }}`
          • *Gross Revenue*: `${{ outputs.validate_and_profile.rows[0].gross_revenue }}`
          • *Avg Order Value*: `${{ outputs.validate_and_profile.rows[0].avg_order_value }}`
          • *Unique Customers*: `{{ outputs.validate_and_profile.rows[0].unique_customers }}`
          • *Artifact*: `{{ outputs.curate_and_export_parquet.outputs[0].uri }}`

    else:
      - id: log_quarantine_alert
        type: io.kestra.plugin.core.log.Log
        description: Log validation failure details for triage and quarantine audit.
        message: "Data Quality Gate breached! Total rows: {{
          outputs.validate_and_profile.rows[0].total_rows }}, Null IDs: {{
          outputs.validate_and_profile.rows[0].null_order_ids }}, Invalid
          Totals: {{ outputs.validate_and_profile.rows[0].invalid_totals }}.
          Ingestion aborted."

  - id: export_curation_metadata
    type: io.kestra.plugin.core.debug.Return
    description: Output execution metrics for downstream data catalog and lineage tracking.
    format: |
      {
        "pipeline": "duckdb-lakehouse-parquet-curation",
        "processed_at": "{{ execution.startDate }}",
        "total_rows_evaluated": {{ outputs.validate_and_profile.rows[0].total_rows }},
        "quality_gate_passed": {{ outputs.validate_and_profile.rows[0].total_rows >= inputs.min_expected_rows and outputs.validate_and_profile.rows[0].null_order_ids == 0 }}
      }