Automate Apache Iceberg Compaction from S3 Events with Spark

id: s3-parquet-spark-iceberg-compaction-trigger
namespace: company.team

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: queue_url
    type: STRING
    displayName: SQS Queue URL
    defaults: "https://sqs.us-east-1.amazonaws.com/123456789012/s3-parquet-landing-queue"
    description: Amazon SQS queue URL receiving S3 ObjectCreated event notifications.

  - id: aws_region
    type: STRING
    displayName: AWS Region
    defaults: "us-east-1"
    description: AWS Region for SQS queue access and S3 object storage operations.

  - id: endpoint_override
    type: STRING
    displayName: Endpoint Override
    defaults: ""
    description: Optional custom endpoint URL for LocalStack, MinIO, or
      S3-compatible test environments.

  - id: s3_bucket
    type: STRING
    displayName: Target S3 Bucket
    defaults: "lakehouse-data"
    description: S3 bucket name containing landed Parquet objects.

  - id: s3_prefix
    type: STRING
    displayName: S3 Key Prefix
    defaults: "warehouse/analytics/customer_orders/"
    description: S3 key prefix where landed Parquet files belong to the target
      Iceberg table.

  - id: iceberg_catalog
    type: STRING
    displayName: Iceberg Catalog Name
    defaults: "lakehouse"
    description: Spark catalog identifier configured for Apache Iceberg.

  - id: iceberg_catalog_impl
    type: STRING
    displayName: Iceberg Catalog Implementation
    defaults: "org.apache.iceberg.rest.RESTCatalog"
    description: Fully qualified class name for the Iceberg catalog implementation
      (e.g., RESTCatalog, GlueCatalog, HadoopCatalog).

  - id: iceberg_catalog_uri
    type: STRING
    displayName: Iceberg Catalog URI
    defaults: "http://iceberg-rest:8181"
    description: Base URI for the Iceberg REST or Nessie catalog service.

  - id: iceberg_warehouse
    type: STRING
    displayName: Iceberg Warehouse Path
    defaults: "s3://lakehouse-data/warehouse"
    description: S3 URI pointing to the warehouse root location where Iceberg
      metadata and data reside.

  - id: iceberg_table
    type: STRING
    displayName: Iceberg Table Identifier
    defaults: "analytics.customer_orders"
    description: Fully qualified table identifier (namespace.table) in the Iceberg catalog.

  - id: small_file_mb
    type: INT
    displayName: Small File Threshold (MB)
    defaults: 64
    description: Maximum file size in megabytes to classify a landed Parquet file as
      a small file.

  - id: min_files
    type: INT
    displayName: Minimum Small Files Threshold
    defaults: 10
    description: Accumulated small file count required to trigger Spark Iceberg compaction.

  - id: min_total_mb
    type: INT
    displayName: Minimum Accumulated Size (MB)
    defaults: 256
    description: Accumulated small file volume in megabytes required to trigger
      Spark Iceberg compaction.

  - id: target_file_mb
    type: INT
    displayName: Target Compaction File Size (MB)
    defaults: 512
    description: Target size in megabytes for compacted data files produced by
      rewrite_data_files.

  - id: max_wait_minutes
    type: INT
    displayName: Maximum Wait Window (Minutes)
    defaults: 120
    description: Time window in minutes after which pending small files trigger
      compaction upon next arrival. Set to 0 to disable time threshold.

  - id: slack_channel
    type: STRING
    displayName: Slack Alert Channel
    defaults: "#lakehouse-alerts"
    description: Slack channel destination for compaction execution reports and
      failure alerts.

variables:
  event_payload: "{{ trigger.data | default('{\"Records\":[]}') }}"
  small_file_limit_bytes: "{{ inputs.small_file_mb * 1024 * 1024 }}"
  min_total_bytes: "{{ inputs.min_total_mb * 1024 * 1024 }}"
  target_file_bytes: "{{ inputs.target_file_mb * 1024 * 1024 }}"

triggers:
  - id: sqs_s3_event_listener
    type: io.kestra.plugin.aws.sqs.RealtimeTrigger
    description: Long-polls Amazon SQS for incoming S3 ObjectCreated event
      notifications in real time.
    queueUrl: "{{ inputs.queue_url }}"
    region: "{{ inputs.aws_region }}"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    endpointOverride: "{{ inputs.endpoint_override != '' ? inputs.endpoint_override : null }}"

tasks:
  - id: process_s3_records
    type: io.kestra.plugin.core.flow.Loop
    description: Inspect each S3 ObjectCreated event record and filter candidate
      small Parquet files.
    values: "{{ (fromJson(trigger.data | default('{\"Records\":[]}')).Records |
      default([])) }}"
    tasks:
      - id: check_eligible_object
        type: io.kestra.plugin.core.flow.If
        description: Verify that the event is ObjectCreated, matches the bucket and
          prefix, and has a .parquet extension.
        condition: "{{ fromJson(item.value).eventName starts with 'ObjectCreated' and
          fromJson(item.value).s3.object.key ends with '.parquet' and
          fromJson(item.value).s3.bucket.name == inputs.s3_bucket and
          fromJson(item.value).s3.object.key starts with inputs.s3_prefix }}"
        then:
          - id: check_small_file_threshold
            type: io.kestra.plugin.core.flow.If
            description: Check if landed Parquet file size is below the configured
              small-file threshold.
            condition: "{{ fromJson(item.value).s3.object.size < (inputs.small_file_mb *
              1024 * 1024) }}"
            then:
              - id: log_candidate_small_file
                type: io.kestra.plugin.core.log.Log
                message: "Small Parquet candidate detected: '{{
                  fromJson(item.value).s3.object.key }}' ({{
                  fromJson(item.value).s3.object.size }} bytes < {{
                  inputs.small_file_mb }} MB)."
            else:
              - id: log_large_file_skipped
                type: io.kestra.plugin.core.log.Log
                message: "Parquet file '{{ fromJson(item.value).s3.object.key }}' ({{
                  fromJson(item.value).s3.object.size }} bytes) exceeds {{
                  inputs.small_file_mb }} MB threshold. Skipping small file
                  accumulation."
        else:
          - id: log_skipped_event
            type: io.kestra.plugin.core.log.Log
            message: "Skipping event {{ fromJson(item.value).eventName }} for key '{{
              fromJson(item.value).s3.object.key | default('unknown') }}': not
              an eligible ObjectCreated Parquet event under prefix '{{
              inputs.s3_prefix }}'."

  - id: update_pending_state
    type: io.kestra.plugin.core.kv.Set
    description: Persist accumulated small file count, byte volume, and oldest event
      timestamp in the KV store.
    key: "pending_compaction_{{ inputs.iceberg_table }}"
    kvType: JSON
    value: |
      {% set current_raw = kv('pending_compaction_' ~ inputs.iceberg_table, errorOnMissing=false) %}
      {% set current_state = current_raw ? fromJson(current_raw) : null %}
      {% set prev_count = current_state ? (current_state.small_file_count | default(0)) : 0 %}
      {% set prev_bytes = current_state ? (current_state.total_small_bytes | default(0)) : 0 %}
      {% set prev_oldest = current_state ? (current_state.oldest_event_time | default('')) : '' %}
      {% set batch_count = (vars.event_payload | jq('[.Records[]? | select((.eventName | startswith("ObjectCreated")) and (.s3.bucket.name == "' ~ inputs.s3_bucket ~ '") and (.s3.object.key | startswith("' ~ inputs.s3_prefix ~ '")) and (.s3.object.key | endswith(".parquet")) and (.s3.object.size < ' ~ vars.small_file_limit_bytes ~ '))] | length') | first) | default(0) %}
      {% set batch_bytes = (vars.event_payload | jq('[.Records[]? | select((.eventName | startswith("ObjectCreated")) and (.s3.bucket.name == "' ~ inputs.s3_bucket ~ '") and (.s3.object.key | startswith("' ~ inputs.s3_prefix ~ '")) and (.s3.object.key | endswith(".parquet")) and (.s3.object.size < ' ~ vars.small_file_limit_bytes ~ ')) | .s3.object.size] | add // 0') | first) | default(0) %}
      {% set batch_oldest = (vars.event_payload | jq('[.Records[]? | select((.eventName | startswith("ObjectCreated")) and (.s3.bucket.name == "' ~ inputs.s3_bucket ~ '") and (.s3.object.key | startswith("' ~ inputs.s3_prefix ~ '")) and (.s3.object.key | endswith(".parquet")) and (.s3.object.size < ' ~ vars.small_file_limit_bytes ~ ')) | .eventTime] | min // empty') | first) | default('') %}
      {% set total_count = prev_count + batch_count %}
      {% set total_bytes = prev_bytes + batch_bytes %}
      {% set oldest_time = prev_oldest != '' ? prev_oldest : (batch_oldest != '' ? batch_oldest : (now() | date("yyyy-MM-dd'T'HH:mm:ss'Z'"))) %}
      {
        "small_file_count": {{ total_count }},
        "total_small_bytes": {{ total_bytes }},
        "oldest_event_time": "{{ oldest_time }}"
      }

  - id: evaluate_compaction_threshold
    type: io.kestra.plugin.core.flow.If
    description: Evaluate whether accumulated small file count or volume exceeds
      compaction thresholds.
    condition: |
      {% set current_raw = kv('pending_compaction_' ~ inputs.iceberg_table, errorOnMissing=false) %}
      {% set current_state = current_raw ? fromJson(current_raw) : null %}
      {% set file_count = current_state ? (current_state.small_file_count | default(0)) : 0 %}
      {% set byte_total = current_state ? (current_state.total_small_bytes | default(0)) : 0 %}
      {{ file_count >= inputs.min_files or byte_total >= vars.min_total_bytes }}
    then:
      - id: run_spark_compaction
        type: io.kestra.plugin.scripts.shell.Commands
        description: Execute Apache Iceberg rewrite_data_files procedure inside an
          Apache Spark container.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: apache/spark:3.5.1
        env:
          AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
          AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
          AWS_REGION: "{{ inputs.aws_region }}"
        commands:
          - |
            /opt/spark/bin/spark-sql               --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.12.262               --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions               --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}=org.apache.iceberg.spark.SparkCatalog               --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.catalog-impl={{ inputs.iceberg_catalog_impl }}               --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.warehouse={{ inputs.iceberg_warehouse }}               {% if inputs.iceberg_catalog_uri != "" %}
              --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.uri={{ inputs.iceberg_catalog_uri }}               {% endif %}
              --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.io-impl=org.apache.iceberg.aws.s3.S3FileIO               {% if inputs.endpoint_override != "" %}
              --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.s3.endpoint={{ inputs.endpoint_override }}               --conf spark.sql.catalog.{{ inputs.iceberg_catalog }}.s3.path-style-access=true               --conf spark.hadoop.fs.s3a.endpoint={{ inputs.endpoint_override }}               --conf spark.hadoop.fs.s3a.path.style.access=true               {% endif %}
              -e "CALL {{ inputs.iceberg_catalog }}.system.rewrite_data_files(table => '{{ inputs.iceberg_table }}', strategy => 'binpack', options => map('target-file-size-bytes', '{{ vars.target_file_bytes }}', 'min-input-files', '{{ inputs.min_files }}'));"
        retry:
          type: constant
          interval: PT15S
          maxAttempts: 2

      - id: reset_pending_state
        type: io.kestra.plugin.core.kv.Delete
        description: Reset the pending compaction tracking state in the KV store after
          successful compaction.
        key: "pending_compaction_{{ inputs.iceberg_table }}"
        errorOnMissing: false

      - id: notify_slack_success
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch compaction execution summary report to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": ":white_check_mark: Iceberg Table Compaction Completed: `{{ inputs.iceberg_table }}`",
            "attachments": [
              {
                "color": "#36a64f",
                "title": "Compaction Summary",
                "fields": [
                  {
                    "title": "Table",
                    "value": "`{{ inputs.iceberg_table }}`",
                    "short": true
                  },
                  {
                    "title": "Execution ID",
                    "value": "<{{ flow.link | default(execution.id) }}|{{ execution.id }}>",
                    "short": true
                  },
                  {
                    "title": "Catalog",
                    "value": "{{ inputs.iceberg_catalog }} ({{ inputs.iceberg_catalog_impl }})",
                    "short": true
                  },
                  {
                    "title": "Target File Size",
                    "value": "{{ inputs.target_file_mb }} MB",
                    "short": true
                  },
                  {
                    "title": "Status",
                    "value": "Compaction successful. KV store pending counters reset to zero.",
                    "short": false
                  }
                ]
              }
            ]
          }
        retry:
          type: constant
          interval: PT5S
          maxAttempts: 3
    else:
      - id: log_pending_status
        type: io.kestra.plugin.core.log.Log
        message: |
          {% set current_raw = kv('pending_compaction_' ~ inputs.iceberg_table, errorOnMissing=false) %}
          {% set current_state = current_raw ? fromJson(current_raw) : null %}
          {% set file_count = current_state ? (current_state.small_file_count | default(0)) : 0 %}
          {% set byte_total = current_state ? (current_state.total_small_bytes | default(0)) : 0 %}
          Pending compaction state for '{{ inputs.iceberg_table }}': {{ file_count }} small file(s), {{ (byte_total / 1024 / 1024) | round(2) }} MB. Thresholds: {{ inputs.min_files }} files or {{ inputs.min_total_mb }} MB. Compaction deferred.

errors:
  - id: notify_slack_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the Spark compaction job fails, while preserving
      pending KV counters for the next retry.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": ":x: Iceberg Compaction Job Failed: `{{ inputs.iceberg_table }}`",
        "attachments": [
          {
            "color": "#E01E5A",
            "title": "Compaction Failure Alert",
            "fields": [
              {
                "title": "Table",
                "value": "`{{ inputs.iceberg_table }}`",
                "short": true
              },
              {
                "title": "Execution ID",
                "value": "<{{ flow.link | default(execution.id) }}|{{ execution.id }}>",
                "short": true
              },
              {
                "title": "KV State Preservation",
                "value": "Pending state preserved in KV store. No file tracking data was lost.",
                "short": false
              }
            ]
          }
        ]
      }
    retry:
      type: constant
      interval: PT5S
      maxAttempts: 3

Automate Apache Iceberg table file compaction using an event-driven architecture powered by Amazon S3, SQS, Kestra KV store, and Apache Spark. In analytical lakehouses with continuous streaming or micro-batch writes, frequent ingestion generates hundreds or thousands of small Parquet files. This small-file accumulation severely degrades query planning and read performance across analytical query engines like Trino, DuckDB, Presto, and Spark.

Rather than running blind, resource-intensive compaction jobs on a fixed schedule, this blueprint acts as an event listener that intercepts Amazon S3 ObjectCreated events from an SQS queue, filters landed Parquet files against configurable size thresholds, accumulates pending small-file metrics in Kestra's persistent KV store, and automatically triggers an isolated Apache Spark container running Iceberg's native rewrite_data_files procedure the moment predefined thresholds are exceeded.

How it works

sqs_s3_event_listener uses io.kestra.plugin.aws.sqs.RealtimeTrigger to listen to an Amazon SQS queue receiving S3 ObjectCreated event notifications in real time.
process_s3_records iterates over each event record in the notification payload using io.kestra.plugin.core.flow.Loop.
check_eligible_object (io.kestra.plugin.core.flow.If) validates that the event is an ObjectCreated event, targets the designated s3_bucket, matches the configured s3_prefix, and ends with .parquet. Non-matching events, non-Parquet files, and deletes are ignored and logged.
check_small_file_threshold identifies files whose size is below small_file_mb (default 64 MB), classifying them as candidate small files requiring compaction.
update_pending_state (io.kestra.plugin.core.kv.Set) increments the table's pending small file count and total byte volume in Kestra's namespace KV store under key pending_compaction_<table_identifier>. To eliminate read-modify-write race conditions when multiple S3 events arrive simultaneously, the flow enforces concurrency: { limit: 1, behavior: QUEUE }, guaranteeing strictly serialized execution slots.
evaluate_compaction_threshold (io.kestra.plugin.core.flow.If) evaluates accumulated metrics against configured thresholds: min_files (e.g., 10 files) or min_total_mb (e.g., 256 MB).
run_spark_compaction (io.kestra.plugin.scripts.shell.Commands with io.kestra.plugin.scripts.runner.docker.Docker) launches an Apache Spark container (apache/spark:3.5.1) with Iceberg runtime packages and executes CALL <catalog>.system.rewrite_data_files(table => ..., strategy => 'binpack', options => map(...)) to bin-pack small files into target-sized data files (e.g., 512 MB). AWS credentials reach the container securely via environment variables from secrets.
reset_pending_state (io.kestra.plugin.core.kv.Delete) deletes the pending counter entry from the KV store upon successful Spark job completion.
notify_slack_success (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) dispatches a structured Slack summary with the table identifier, execution ID, catalog details, and compaction confirmation.
notify_slack_failure in the flow-level errors: handler catches any container or compaction failures, posts an incident alert to Slack, and intentionally preserves the KV state so pending small files are not forgotten and will be compacted on the next execution cycle.

What you get

Autonomous event-driven lakehouse maintenance triggered by real S3 landing telemetry.
Safe small-file bin-packing using Apache Iceberg's native rewrite_data_files Spark procedure.
Atomic, race-free state tracking via Kestra's namespace KV store and execution concurrency queueing.
Zero persistent Spark cluster infrastructure overhead: Spark runs inside an ephemeral Docker container only when compaction thresholds are met.
Resilient failure recovery: failed compaction jobs preserve accumulated KV counts, alert on-call engineers via Slack, and fail cleanly without data loss.

Who it's for

Lakehouse Engineers and Platform Engineers managing Apache Iceberg tables on Amazon S3.
Data Engineers looking to eliminate small-file query degradation from streaming ingestion.
Analytics Platform teams automating Day-2 table maintenance with declarative infrastructure.

Why orchestrate this with Kestra

Running compaction on fixed cron schedules is inefficient: jobs fire during quiet periods when no files arrived, or fail to keep up during heavy ingestion bursts. Pure script-based consumers lack durable execution state, retry logic, concurrency isolation, and container management.

Orchestrating this workflow with Kestra combines event-driven triggers, persistent KV state storage, strict execution serialization (limit: 1, behavior: QUEUE), containerized Spark execution via Docker task runners, and failure routing in a single declarative YAML definition.

Prerequisites

An Amazon SQS queue configured to receive s3:ObjectCreated:* notifications from your target S3 bucket.
An Apache Iceberg catalog (REST Catalog, AWS Glue, or Hadoop Catalog) reachable from the Kestra worker environment.
A Kestra worker running with Docker task runner enabled to execute the Apache Spark container.
A Slack Incoming Webhook URL for compaction reports and incident alerts.

Secrets

AWS_ACCESS_KEY_ID: AWS access key with permissions to read from the SQS queue and read/write to the target S3 bucket and warehouse prefix.
AWS_SECRET_ACCESS_KEY: AWS secret access key corresponding to the access key ID.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL used to send compaction summaries and failure notifications.

Quick start

Configure the three required secrets in Kestra: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and SLACK_WEBHOOK_URL.
Configure your S3 bucket event notifications to publish All object create events for prefix warehouse/analytics/customer_orders/ to your Amazon SQS queue.
Update the flow inputs: set queue_url, s3_bucket, s3_prefix, iceberg_catalog_uri, and iceberg_table to point to your lakehouse environment.
Save the flow and ensure the sqs_s3_event_listener trigger is active. Land Parquet files in S3 and observe automated small file accumulation and compaction in the Kestra execution UI.

How to extend

Partition filtering: Add a where clause option to the rewrite_data_files Spark procedure (e.g., where => 'order_date >= current_date() - interval 7 days') to constrain compaction to recently landed partitions.
Snapshot expiration: Chain a follow-up task executing CALL <catalog>.system.expire_snapshots(table => ...) to remove obsolete snapshots and prune unreferenced manifests after compaction.
Multi-table routing: Expand the record filtering logic to handle multiple table prefixes by mapping prefix paths to distinct table identifiers and isolated KV storage keys.

S3 Parquet Spark Iceberg Compaction Trigger

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

S3 Parquet Spark Iceberg Compaction Trigger

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

OceanBase Major Compaction Stall & Auto-Resume Sentinel

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

NFS New File Extension Compliance Gate

Detect and Resume Stuck Paused NSQ Channels with Accumulating Backlog

Object Storage Integrity and Streaming Replica Parity Audit

Search OpenAlex for Related Scholarly Works

New to Kestra?