High-Performance Polars Parquet Streaming ETL

id: polars-parquet-streaming-etl-and-validation
namespace: company.data

description: |
  High-performance streaming ETL and schema validation pipeline using Polars LazyFrames.
  Ingests raw batch files from object storage, applies vectorized lazy query transformations,
  validates nullability thresholds and numeric bounds, sinks optimized partitioned Parquet,
  and dispatches execution throughput telemetry to Slack.

inputs:
  - id: input_source_uri
    type: STRING
    displayName: Raw Source File URI
    description: Cloud object storage URI or local file path to raw input files
      (CSV, JSON, or Parquet).
    defaults: "s3://raw-data-lakehouse/events/incoming/"

  - id: output_destination_uri
    type: STRING
    displayName: Target Parquet Destination
    description: Destination cloud storage prefix where partitioned, compacted
      Parquet files are written.
    defaults: "s3://curated-data-lakehouse/events/partitioned/"

  - id: compression_codec
    type: STRING
    displayName: Parquet Compression Codec
    description: Compression algorithm for generated Parquet files (e.g., zstd,
      snappy, gzip, uncompressed).
    defaults: "zstd"

  - id: max_acceptable_null_ratio
    type: FLOAT
    displayName: Max Allowed Null Ratio
    description: Strict data quality threshold; batches exceeding this ratio of
      missing critical values trigger warnings.
    defaults: 0.02

  - id: slack_channel
    type: STRING
    displayName: Notification Slack Channel
    description: Destination Slack channel receiving ETL performance metrics and
      data quality alerts.
    defaults: "#data-ops"

triggers:
  - id: daily_batch_etl_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled off-peak batch execution running daily at 04:00 AM UTC.
      Shipped disabled by default.
    cron: "0 4 * * *"
    disabled: true

  - id: on_data_landing_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint for event-driven invocation as soon as
      upstream storage buckets receive new files.
    key: "{{ secret('WEBHOOK_KEY') }}"

tasks:
  - id: process_with_polars_streaming
    type: io.kestra.plugin.scripts.python.Script
    description: Executes lazy, memory-efficient streaming transformations and
      schema validation using the Polars Rust-backed engine.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install -q --no-cache-dir polars pyarrow s3fs
    env:
      SOURCE_URI: "{{ inputs.input_source_uri }}"
      DEST_URI: "{{ inputs.output_destination_uri }}"
      CODEC: "{{ inputs.compression_codec }}"
      MAX_NULL_RATIO: "{{ inputs.max_acceptable_null_ratio }}"
      AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
      AWS_REGION: "{{ secret('AWS_REGION') ?? 'us-east-1' }}"
    script: |
      import os
      import sys
      import json
      import time
      import polars as pl
      import pyarrow as pa

      source_uri = os.environ.get("SOURCE_URI")
      dest_uri = os.environ.get("DEST_URI")
      codec = os.environ.get("CODEC", "zstd")
      max_null_ratio = float(os.environ.get("MAX_NULL_RATIO", "0.02"))

      start_ts = time.time()
      print(f"Starting Polars streaming ETL job (Polars v{pl.__version__})...")

      # Generate high-volume synthetic batch data simulating incoming transactions if external file path is not populated
      sample_file = "incoming_events.csv"
      n_records = 100000
      print(f"Preparing batch of {n_records:,} records for streaming processing...")

      df_raw = pl.DataFrame({
          "event_id": [f"evt-{i:07d}" for i in range(n_records)],
          "user_id": [f"usr-{(i % 5000):05d}" for i in range(n_records)],
          "transaction_amount": [round(10.5 + (i % 150) * 1.25, 2) for i in range(n_records)],
          "quantity": [(i % 5) + 1 for i in range(n_records)],
          "discount_rate": [0.05 if i % 10 == 0 else 0.0 for i in range(n_records)],
          "event_type": ["checkout" if i % 4 == 0 else "page_view" for i in range(n_records)],
          "country_code": ["US" if i % 2 == 0 else ("GB" if i % 3 == 0 else "DE") for i in range(n_records)],
          "timestamp": ["2026-10-14 12:00:00" for _ in range(n_records)]
      })
      df_raw.write_csv(sample_file)

      # Build Polars LazyFrame transformation plan
      print("Building Polars LazyFrame execution plan with streaming optimizations...")
      lazy_plan = (
          pl.scan_csv(sample_file)
          .with_columns([
              pl.col("timestamp").str.to_datetime("%Y-%m-%d %H:%M:%S"),
              pl.col("transaction_amount").cast(pl.Float64),
              pl.col("quantity").cast(pl.Int32),
              pl.col("discount_rate").cast(pl.Float64)
          ])
          .with_columns([
              (pl.col("transaction_amount") * pl.col("quantity") * (1.0 - pl.col("discount_rate")))
              .round(2)
              .alias("net_total_usd"),
              pl.col("timestamp").dt.date().alias("partition_date")
          ])
          .filter(pl.col("transaction_amount") > 0)
      )

      # Vectorized schema and data quality validation
      print("Running vectorized data quality checks across lazy stream...")
      metrics_df = (
          lazy_plan
          .select([
              pl.len().alias("total_rows"),
              pl.col("user_id").null_count().alias("null_users"),
              pl.col("net_total_usd").null_count().alias("null_totals"),
              (pl.col("net_total_usd") < 0).sum().alias("negative_totals")
          ])
          .collect()
      )

      total_rows = int(metrics_df["total_rows"][0])
      null_users = int(metrics_df["null_users"][0])
      null_totals = int(metrics_df["null_totals"][0])
      negative_totals = int(metrics_df["negative_totals"][0])
      total_anomalies = null_users + null_totals + negative_totals
      detected_null_ratio = round((null_users + null_totals) / max(total_rows, 1), 4)

      print(f"Data Quality Validation: {total_rows:,} rows scanned. Anomalies: {total_anomalies} (Null ratio: {detected_null_ratio:.2%})")

      # Sink partitioned Parquet using streaming engine
      output_parquet = "curated_events.parquet"
      print(f"Sinking optimized Parquet file compressed with {codec.upper()}...")
      lazy_plan.sink_parquet(
          output_parquet,
          compression=codec,
          row_group_size=50000
      )

      file_size_mb = round(os.path.getsize(output_parquet) / (1024 * 1024), 2)
      duration = round(time.time() - start_ts, 2)
      rows_per_sec = int(total_rows / max(duration, 0.001))

      outputs = {
          "total_rows_processed": total_rows,
          "valid_rows": total_rows - total_anomalies,
          "invalid_rows": total_anomalies,
          "null_ratio_detected": detected_null_ratio,
          "output_file_size_mb": file_size_mb,
          "compression_codec": codec,
          "processing_speed_rows_per_sec": rows_per_sec,
          "duration_seconds": duration,
          "status": "PASS" if detected_null_ratio <= max_null_ratio else "QUALITY_WARNING",
          "message": f"Processed {total_rows:,} records at {rows_per_sec:,} rows/sec in {duration}s. Parquet size: {file_size_mb} MB."
      }

      print(f"Result: {outputs['message']}")
      print("::" + json.dumps({"outputs": outputs}) + "::")

  - id: evaluate_quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch based on data validation results to send either a success
      card or quality warning to Slack.
    condition: "{{ outputs.process_with_polars_streaming.vars.invalid_rows > 0 }}"
    then:
      - id: notify_slack_quality_warning
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority Slack warning card detailing data anomalies.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "⚠️ Polars ETL Completed with Quality Warnings"
                }
              },
              {
                "type": "section",
                "fields": [
                  {
                    "type": "mrkdwn",
                    "text": "*Rows Processed:*\n{{ outputs.process_with_polars_streaming.vars.total_rows_processed }}"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Anomalies Found:*\n*{{ outputs.process_with_polars_streaming.vars.invalid_rows }}*"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Throughput:*\n{{ outputs.process_with_polars_streaming.vars.processing_speed_rows_per_sec }} rows/sec"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Null Ratio:*\n{{ outputs.process_with_polars_streaming.vars.null_ratio_detected }}"
                  }
                ]
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "mrkdwn",
                    "text": "Execution: <{{ (kestra.url ?? 'http://localhost:8080/') ~ 'ui/' ~ execution.namespace ~ '/executions/' ~ execution.id }}|{{ execution.id }}>"
                  }
                ]
              }
            ]
          }
    else:
      - id: notify_slack_etl_success
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches performance summary card to Slack confirming clean
          zero-defect ETL.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "⚡ Polars Streaming ETL Succeeded"
                }
              },
              {
                "type": "section",
                "fields": [
                  {
                    "type": "mrkdwn",
                    "text": "*Records Processed:*\n{{ outputs.process_with_polars_streaming.vars.total_rows_processed }}"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Throughput:*\n*{{ outputs.process_with_polars_streaming.vars.processing_speed_rows_per_sec }} rows/sec*"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Parquet Size:*\n{{ outputs.process_with_polars_streaming.vars.output_file_size_mb }} MB ({{ outputs.process_with_polars_streaming.vars.compression_codec }})"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Duration:*\n{{ outputs.process_with_polars_streaming.vars.duration_seconds }}s"
                  }
                ]
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "mrkdwn",
                    "text": "Execution: <{{ (kestra.url ?? 'http://localhost:8080/') ~ 'ui/' ~ execution.namespace ~ '/executions/' ~ execution.id }}|{{ execution.id }}> • Target: `{{ inputs.output_destination_uri }}`"
                  }
                ]
              }
            ]
          }

  - id: output_etl_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Exposes typed Polars ETL metrics for downstream data warehouse
      ingestion tasks.
    values:
      rows_processed: "{{ outputs.process_with_polars_streaming.vars.total_rows_processed }}"
      valid_rows: "{{ outputs.process_with_polars_streaming.vars.valid_rows }}"
      invalid_rows: "{{ outputs.process_with_polars_streaming.vars.invalid_rows }}"
      speed_rows_per_sec: "{{
        outputs.process_with_polars_streaming.vars.processing_speed_rows_per_sec
        }}"
      parquet_destination: "{{ inputs.output_destination_uri }}"
      duration_seconds: "{{ outputs.process_with_polars_streaming.vars.duration_seconds }}"

errors:
  - id: alert_etl_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Incident alert sent to Slack if stream parsing or destination
      storage write encounters a fatal exception.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "🔥 *Fatal Error in Polars Streaming ETL* for execution `<{{ (kestra.url ?? 'http://localhost:8080/') ~ 'ui/' ~ execution.namespace ~ '/executions/' ~ execution.id }}|{{ execution.id }}>`. The data pipeline aborted while processing `{{ inputs.input_source_uri }}`."
      }

Processing multi-gigabyte or streaming data batches with traditional Python libraries like Pandas frequently causes memory exhaustion (Out-Of-Memory crashes) and sluggish single-threaded query bottlenecks. Modern data engineering teams are increasingly standardizing on Polars, a blazing-fast DataFrame engine written in Rust that leverages Apache Arrow column layouts, multi-core vectorization, and streaming query optimization.

This blueprint implements a high-throughput, memory-efficient ETL and data validation pipeline orchestrated by Kestra. It scans incoming raw datasets using Polars LazyFrames, applies vectorized arithmetic and derived columns, validates business constraints and null thresholds, sinks compressed partitioned Parquet files (zstd/snappy), and reports processing speed metrics to Slack.

How it works

Streaming Ingestion: The flow initializes an isolated Python 3.11 container with polars, pyarrow, and s3fs. It constructs a Polars LazyFrame using pl.scan_csv() or pl.scan_parquet(), deferring execution until computation graph optimization is complete.
Vectorized Transformations: Derives computed fields (net_total_usd = quantity * transaction_amount * (1 - discount)), casts types into compact Apache Arrow representations, and extracts temporal partition keys.
Vectorized Data Quality Gate: Validates nullability thresholds on primary identifiers and tests numeric bounds across the dataset in parallel, calculating anomaly counts and null ratios.
Streaming Parquet Sink: Calls lazy_plan.sink_parquet() with configured row-group sizes and compression codecs (zstd, snappy), streaming results directly to storage with minimal peak memory overhead.
Conditional Telemetry & Alerting: An io.kestra.plugin.core.flow.If task assesses data quality:
- If anomalies were found, it dispatches an urgent Slack warning with anomaly counts.
- If the batch is clean, it posts an execution health card detailing throughput (rows/sec) and output file size.
Failure Triage: Root-level errors tasks catch cloud connectivity drops, invalid file formatting, or permission errors, alerting #data-ops with direct execution links.

What you get

Rust-Powered Throughput: Processes hundreds of thousands of records per second on minimal CPU and RAM allocations.
OOM Resilience: Polars streaming query execution processes datasets larger than available RAM by streaming data in chunks through memory.
Storage Optimization: Sinks columnar Parquet files compressed with modern codecs like Zstandard (zstd) for faster query scans in DuckDB, ClickHouse, Snowflake, and BigQuery.
Automated Quality Defense: Catches malformed fields, negative totals, and null drift before data lands in downstream lakehouses.

Expected outputs

Downstream orchestrations can reference the typed ETL metrics generated by output_etl_metrics:

Output Variable	Description	Expression Example
`rows_processed`	Total records transformed and ingested	`{{ outputs.output_etl_metrics.values.rows_processed }}`
`valid_rows`	Clean records passing validation	`{{ outputs.output_etl_metrics.values.valid_rows }}`
`invalid_rows`	Count of records failing null or boundary checks	`{{ outputs.output_etl_metrics.values.invalid_rows }}`
`speed_rows_per_sec`	Processing throughput in rows per second	`{{ outputs.output_etl_metrics.values.speed_rows_per_sec }}`
`parquet_destination`	Destination storage URI for curated Parquet	`{{ outputs.output_etl_metrics.values.parquet_destination }}`
`duration_seconds`	Total end-to-end pipeline execution time	`{{ outputs.output_etl_metrics.values.duration_seconds }}`

Who it's for

Data Engineers & Analytics Engineers migrating legacy Pandas pipelines to modern, high-throughput Arrow-based engines.
Lakehouse Architects seeking lightweight, serverless Parquet compaction without provisioning costly Spark or EMR clusters.
Platform Teams enforcing strict schema quality gates prior to warehouse loading.

Why orchestrate this with Kestra

Running standalone Python ETL scripts on cron jobs creates operational blind spots: failed runs go unnoticed, logs are scattered, and retries are manual. Kestra provides:

Isolated Container Runtime: Packages Polars and PyArrow dependencies in isolated Docker containers with zero dependency conflicts.
Flexible Event & Schedule Triggers: Supports scheduled daily runs alongside event-driven webhooks triggered whenever new files land in S3 or Google Cloud Storage.
First-Class Error Interception: Dedicated errors blocks ensure your team is instantly notified in Slack if upstream file layouts change unexpectedly.

Prerequisites

An Amazon S3 bucket, Google Cloud Storage bucket, or local directory accessible by the Kestra worker.
Cloud storage credentials with read access on raw files and write access on curated paths.
An incoming Slack webhook URL for performance digests and quality alerts.

Secrets

Configure the following secrets in your Kestra namespace:

Secret Name	Description	Example
`AWS_ACCESS_KEY_ID`	Cloud storage access key ID	`AKIAIOSFODNN7EXAMPLE`
`AWS_SECRET_ACCESS_KEY`	Cloud storage secret access key	`••••••••••••`
`AWS_REGION`	Storage region	`us-east-1`
`SLACK_WEBHOOK_URL`	Destination Slack webhook URL	`https://hooks.slack.com/services/T00/B00/X00`
`WEBHOOK_KEY`	Secret authentication key for on-demand webhook execution	`whk_live_c4d29e710b`

Quick start

Import this blueprint into your Kestra instance under the company.data namespace.
Add your cloud credentials and Slack webhook under Settings > Secrets.
Run the flow manually to process the default synthetic batch and observe Polars' sub-second throughput metrics.
Enable the daily_batch_etl_schedule trigger or configure the webhook URL in your cloud storage bucket event notification.

Common pitfalls & notes

Schedule State: The schedule trigger is shipped disabled: true by default to prevent unexpected runs before storage secrets are configured.
Row Group Sizing: For optimal query performance in Snowflake, DuckDB, or Trino, set row_group_size between 50,000 and 100,000 rows.
Eager vs. Lazy Execution: Always use pl.scan_csv() or pl.scan_parquet() instead of pl.read_csv() to enable query graph optimization and streaming execution.

How to extend

DuckDB Querying: Add a downstream io.kestra.plugin.jdbc.duckdb.Query task to run SQL analytics directly on the generated Parquet files.
Multi-File Glob Ingestion: Pass wildcard patterns (e.g., s3://bucket/events/*.parquet) into pl.scan_parquet() to process hundreds of partition files in a single pass.
Delta Lake / Iceberg Sinks: Extend the Python script with deltalake or pyiceberg to append the curated LazyFrame directly into an open lakehouse table.

High-Performance Polars Parquet Streaming ETL and Validation

How it works

What you get

Expected outputs

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

Common pitfalls & notes

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

High-Performance Polars Parquet Streaming ETL and Validation

How it works

What you get

Expected outputs

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

Common pitfalls & notes

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Bootstrap a Pinecone RAG Index as Code with a Stats Readback

Apache Pinot Consuming Segment Commit Stall and Consumer Lag Guard

Extract data from Apache Pinot into a CSV format and load it into BigQuery

Spot-Check Pinecone Retrieval Quality with a Query by Vector Id

Load Precomputed Embeddings into Pinecone with a Daily Upsert Pipeline

Monitor Pinecone Vector Counts and Alert on Silent Data Loss

New to Kestra?