Lakehouse ETL on Kubernetes: an introduction to DataFlow Operator

#opensource #kubernetes #data #dataengineering

uilding a lakehouse means continuously landing data in Apache Iceberg: Kafka events into bronze, OLTP increments, CDC into silver. The usual stack for that is Airflow + Spark plus a pile of glue code. For the common path “read → lightly reshape → write to Iceberg,” that stack is often overkill.

DataFlow Operator is a Kubernetes operator that declares ingest as a CRD: source → transformations → sink. It runs the processor, handles restarts and checkpoints, and—for batch jobs—can kick off post-load steps such as Spark on Iceberg tables.

Below is a short refresher on ETL, then three practical ways to land data in a lakehouse: streaming Extract, minimal streaming ETL, and batch ELT with Spark after the load.

What ETL is (and ELT / streaming next to it)

ETL means Extract, Transform, Load:

Extract — pull data from a source (Kafka, CDC, polling SQL).
Transform — reshape it (filter, flatten, mask fields, rename keys).
Load — write to a sink. In a lakehouse that is usually an Apache Iceberg table via a REST Catalog (Polaris, Lakekeeper, iceberg-rest) or Nessie.

Two nearby patterns:

Streaming ETL — the same loop, continuously: messages land in bronze/silver without waiting for end-of-day.
ELT — Load into Iceberg first (often bronze), then run heavy transforms with Spark/Trino/SQL after the load.

On Kubernetes it helps when streaming and batch share the same manifest shape—only the run mode changes. The sink in the examples below is the iceberg connector (REST Catalog). For a Nessie catalog with branches, use the separate nessie type; the Iceberg table model is the same.

Where DataFlow fits

One pipeline runtime pattern, two kinds:

Kind	Mode	When to use it
DataFlow	Continuous Deployment	Steady stream into Iceberg (Kafka, CDC)
DataFlowCron	CronJob + per-tick Job	Schedule and/or Spark/SQL after a successful load (`triggers`)

Pipeline model:

source → [transformations...] → sink (iceberg)
         └─ optional errors (DLQ)

The operator owns pod lifecycle, at-least-once delivery, checkpoints for polling sources, and Secrets integration. Heavy compute (Spark, dbt, an Airflow DAG) stays outside the processor: via DataFlowCron triggers or a separate downstream job on lakehouse tables.

Docs: dataflow-operator.github.io/docs.

Repo: github.com/dataflow-operator/dataflow.

Option 1. Streaming Extract into Iceberg

Goal: continuously consume Kafka and append into bronze Iceberg with no transforms.

Pure Extract + Load: the consumer loop, ack, and Deployment restarts are the operator’s job. You only declare source and sink.

apiVersion: dataflow.dataflow.io/v1
kind: DataFlow
metadata:
  name: kafka-to-iceberg
spec:
  source:
    type: kafka
    config:
      brokers:
        - kafka:9092
      topic: input-topic
      consumerGroup: dataflow-group
  sink:
    type: iceberg
    config:
      catalogURI: "https://iceberg-catalog.example.com"
      warehouse: main
      namespace: bronze
      table: events
      batchSize: 100
      autoCreateTable: true
      authenticationType: BEARER
      tokenSecretRef:
        name: iceberg-catalog
        key: token

Use this when:

you need a firehose from a topic into an Iceberg table;
schema cleanup can wait for Spark/Trino;
ingest-time transforms are not required yet.

The same pattern works for CDC (postgresql-cdc, Debezium on Kafka)—Extract stays streaming; only the source type changes; the sink remains Iceberg.

Option 2. Minimal streaming ETL with transformers

Goal: lightly reshape JSON in flight—flatten an array, add a timestamp, drop noise, mask PII—then append to Iceberg.

DataFlow supports an in-process transform chain (JSONPath via gjson): flatten, timestamp, filter, select, remove, mask, snakeCase / camelCase, debeziumUnwrap, router, and more.

apiVersion: dataflow.dataflow.io/v1
kind: DataFlow
metadata:
  name: stock-to-iceberg
spec:
  source:
    type: kafka
    config:
      brokers:
        - kafka:9092
      topic: stock-topic
      consumerGroup: dataflow-group
  transformations:
    - type: flatten
      config:
        field: rowsStock
    - type: timestamp
      config:
        fieldName: created_at
  sink:
    type: iceberg
    config:
      catalogURI: "https://iceberg-catalog.example.com"
      warehouse: main
      namespace: silver
      table: stock_items
      batchSize: 100
      autoCreateTable: true
      authenticationType: BEARER
      tokenSecretRef:
        name: iceberg-catalog
        key: token

This is not a Spark/Flink replacement for lakehouse joins and heavy analytics. For “unwrap an envelope → keep the fields you need → mask a card number → land in Iceberg,” you often do not need a separate compute cluster at ingest time.

A typical CDC path: Debezium on Kafka → debeziumUnwrap → select / mask → append to a silver Iceberg table.

Option 3. Batch ELT: DataFlowCron + Spark after Iceberg load

Goal: on a schedule, pull an OLTP increment (polling SQL), land it in bronze Iceberg, then start Spark for heavy transforms on lakehouse tables.

DataFlow covers EL (Extract + Load into Iceberg); Spark owns T after the Job succeeds. DataFlowCron exposes ordered triggers that start only when the processor Job completes successfully (JobComplete).

Important: for post-load triggers, use a polling source (postgresql, clickhouse, trino, nessie / iceberg). Kafka under cron is streaming—the Job often never reaches “source exhausted,” so triggers never run.

apiVersion: dataflow.dataflow.io/v1
kind: DataFlowCron
metadata:
  name: orders-hourly-elt
spec:
  schedule: "0 * * * *"
  concurrencyPolicy: Forbid
  checkpointSyncOnAck: true
  source:
    type: postgresql
    config:
      connectionStringSecretRef:
        name: source-db
        key: url
      table: orders
      changeTrackingColumn: updated_at
      orderByColumn: id
      pollInterval: 30
      readBatchSize: 1000
  sink:
    type: iceberg
    config:
      catalogURI: "https://iceberg-catalog.example.com"
      warehouse: main
      namespace: bronze
      table: orders
      batchSize: 100
      autoCreateTable: true
      authenticationType: BEARER
      tokenSecretRef:
        name: iceberg-catalog
        key: token
  triggers:
    - name: start-spark
      image: bitnami/kubectl:latest
      command: ["kubectl"]
      args: ["apply", "-f", "/manifests/spark-application.yaml"]

How to read this:

Once an hour, the CronJob starts a processor.
The processor reads the PostgreSQL increment from the checkpoint (updated_at, id) and appends into bronze.orders (Iceberg).
The source is exhausted → the Job succeeds → the trigger applies a SparkApplication (or runs spark-submit from its image)—for example bronze → silver/gold.

Spark is not a first-class connector here; it is a generic post-step: any container with image / command / args. Bake the SparkApplication manifest into the trigger image (the triggers API has no volumeMounts). The same hook can start an Airflow DAG or refresh Trino/BI.

checkpointSyncOnAck on cron helps survive at-least-once retries on Extract; merge/idempotency in the lakehouse is usually handled in the Spark job against Iceberg (MERGE / partition overwrite).

Which option to pick

Need	Kind	Pattern
Continuous land into Iceberg, no reshaping	`DataFlow`	Extract (option 1)
Light JSON reshape in flight → Iceberg	`DataFlow` + `transformations`	Streaming ETL (option 2)
Schedule + Spark after Iceberg load	`DataFlowCron` + `triggers`	ELT (option 3)
Steady stream and post-steps	two resources or an intermediate topic	DataFlow → Spark/orchestrator separately

Product boundaries to keep in mind:

transforms are in-process, per message/row—not distributed lakehouse joins;
triggers are an ordered Job chain, not a full DAG orchestrator;
delivery is at-least-once; strong idempotency on Iceberg is usually achieved in Spark (MERGE / overwrite partition).

Next steps

Repository and issues: github.com/dataflow-operator/dataflow
Documentation: dataflow-operator.github.io/docs
Helm quick start: Getting started
Iceberg connector: Connectors
Transformations: Transformations
Cron and post-load: DataFlowCron triggers

If you already run “simple” ingest as scripts—or a heavy Airflow DAG just to move Kafka into Iceberg—consider declaring it as a DataFlow or DataFlowCron and keep the orchestrator for the parts that truly need a DAG over lakehouse tables.