Polars and DuckDB High-Performance Parquet Lakehouse Pipeline

id: polars-duckdb-parquet-data-transformation-pipeline
namespace: company.analytics

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: input_dataset_url
    type: STRING
    displayName: Input Dataset URL
    required: true
    defaults: "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/titanic.csv"
    description: "Public HTTP/HTTPS source CSV file to ingest and process."

  - id: output_s3_bucket
    type: STRING
    displayName: Output S3 Bucket
    required: true
    defaults: "analytics-curated-parquet-lakehouse"
    description: "Destination S3 bucket to publish curated dimensional Parquet tables."

  - id: output_s3_region
    type: STRING
    displayName: AWS S3 Region
    required: true
    defaults: "us-east-1"
    description: "AWS region hosting the lakehouse bucket."

  - id: min_row_count
    type: INT
    displayName: Minimum Record Count Barrier
    required: true
    defaults: 500
    description: "Quality barrier threshold: fails the pipeline if the processed
      dataset contains fewer valid rows."

tasks:
  - id: log_pipeline_start
    type: io.kestra.plugin.core.log.Log
    message: "⚡ Initiating Polars high-performance data transformation and DuckDB
      quality audit on '{{ inputs.input_dataset_url }}'"

  - id: download_source_data
    type: io.kestra.plugin.core.http.Download
    description: "Ingest source raw dataset via HTTP stream into Kestra storage"
    uri: "{{ inputs.input_dataset_url }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3

  - id: polars_transform_and_clean
    type: io.kestra.plugin.scripts.python.Script
    description: "Execute vectorized Polars data cleaning, schema casting, and
      Parquet generation in Docker"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    dependencies:
      - polars==1.8.2
      - pyarrow==17.0.0
      - kestra
    inputFiles:
      raw_source.csv: "{{ outputs.download_source_data.uri }}"
    script: |
      import polars as pl
      from kestra import Kestra

      print("Loading raw CSV with Polars scan...")
      df = pl.read_csv("raw_source.csv")

      initial_count = len(df)
      print(f"Read {initial_count} initial records with columns: {df.columns}")

      cleaned_df = (
          df.filter(pl.col("age").is_not_null())
          .with_columns([
              pl.col("fare").cast(pl.Float64).round(2),
              pl.col("age").cast(pl.Float64),
              pl.when(pl.col("survived") == 1)
                .then(pl.lit("SURVIVED"))
                .otherwise(pl.lit("PERISHED"))
                .alias("status")
          ])
      )

      valid_rows = len(cleaned_df)
      output_filename = "curated_transformed.parquet"
      cleaned_df.write_parquet(output_filename, compression="zstd")
      print(f"Transformed dataset written to {output_filename}: {valid_rows} clean rows.")

      Kestra.outputs({
          "initial_count": initial_count,
          "valid_rows": valid_rows,
          "parquet_file": output_filename
      })
    outputFiles:
      - "*.parquet"

  - id: duckdb_quality_and_contract_audit
    type: io.kestra.plugin.scripts.python.Script
    description: "Run in-memory DuckDB SQL data contracts and anomaly detection
      against the generated Parquet file"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    dependencies:
      - duckdb==1.1.0
      - kestra
    inputFiles:
      curated_transformed.parquet: "{{
        outputs.polars_transform_and_clean.outputFiles['curated_transformed.par\
        quet'] }}"
    script: |
      import duckdb
      from kestra import Kestra

      con = duckdb.connect(database=":memory:")

      # 1. Verify schema contract
      schema_df = con.execute("DESCRIBE SELECT * FROM 'curated_transformed.parquet'").fetchall()
      print("Parquet Schema Verified by DuckDB:", schema_df)

      # 2. Check for null or negative values in critical fields
      anomalies = con.execute("""
          SELECT count(*) FROM 'curated_transformed.parquet'
          WHERE age < 0 OR fare < 0 OR status IS NULL
      """).fetchone()[0]

      total_records = con.execute("SELECT count(*) FROM 'curated_transformed.parquet'").fetchone()[0]
      print(f"DuckDB Audit: {total_records} records, {anomalies} anomalies.")

      Kestra.outputs({
          "total_records": total_records,
          "anomalies_detected": anomalies,
          "passed_audit": (anomalies == 0 and total_records >= {{ inputs.min_row_count }})
      })

  - id: check_data_quality_gate
    type: io.kestra.plugin.core.flow.If
    description: "Halt pipeline if DuckDB detects schema violations or insufficient records"
    condition: "{{ not outputs.duckdb_quality_and_contract_audit.vars.passed_audit }}"
    then:
      - id: fail_quality_audit
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "Quality gate failed: DuckDB found {{
          outputs.duckdb_quality_and_contract_audit.vars.anomalies_detected }}
          anomalies and {{
          outputs.duckdb_quality_and_contract_audit.vars.total_records }} valid
          rows (minimum required: {{ inputs.min_row_count }})."

  - id: upload_to_lakehouse_s3
    type: io.kestra.plugin.aws.s3.Upload
    description: "Publish verified Parquet dataset to AWS S3 data lakehouse"
    from: "{{
      outputs.polars_transform_and_clean.outputFiles['curated_transformed.parqu\
      et'] }}"
    bucket: "{{ inputs.output_s3_bucket }}"
    key: "curated-lakehouse/{{ execution.startDate | date('yyyy-MM-dd')
      }}/dataset_{{ execution.id }}.parquet"
    region: "{{ inputs.output_s3_region }}"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"

  - id: record_lakehouse_catalog
    type: io.kestra.plugin.core.kv.Set
    description: "Update Kestra KV store with latest dataset publication coordinates"
    key: "curated_dataset_latest"
    value: |
      {
        "s3_path": "s3://{{ inputs.output_s3_bucket }}/curated-lakehouse/{{ execution.startDate | date('yyyy-MM-dd') }}/dataset_{{ execution.id }}.parquet",
        "record_count": {{ outputs.duckdb_quality_and_contract_audit.vars.total_records }},
        "timestamp": "{{ execution.startDate }}",
        "execution_id": "{{ execution.id }}"
      }

  - id: notify_slack_publication
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Post publication confirmation to data engineering Slack channel"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "blocks": [
          {
            "type": "header",
            "text": {
              "type": "plain_text",
              "text": "📊 Lakehouse Parquet Dataset Published",
              "emoji": true
            }
          },
          {
            "type": "section",
            "fields": [
              {
                "type": "mrkdwn",
                "text": "*Records:* `{{ outputs.duckdb_quality_and_contract_audit.vars.total_records }}`"
              },
              {
                "type": "mrkdwn",
                "text": "*Destination:* `s3://{{ inputs.output_s3_bucket }}`"
              },
              {
                "type": "mrkdwn",
                "text": "*Quality Check:* `Passed (0 anomalies)`"
              },
              {
                "type": "mrkdwn",
                "text": "*Execution:* `{{ execution.id }}`"
              }
            ]
          }
        ]
      }

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Dead-letter alert dispatched on data transformation or upload error"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "blocks": [
          {
            "type": "header",
            "text": {
              "type": "plain_text",
              "text": "🚨 Lakehouse ETL Pipeline FAILED",
              "emoji": true
            }
          },
          {
            "type": "section",
            "text": {
              "type": "mrkdwn",
              "text": "Data transformation failed on execution `{{ execution.id }}` for source `{{ inputs.input_dataset_url }}`."
            }
          }
        ]
      }