Aller au contenu principal

Streaming & CDC Ingestion with Kafka

Architecture Overview​

Ilum provides a natural foundation for real-time streaming workloads. Apache Kafka is already a core dependency of the ilum platform, used internally for service communication. This means every ilum deployment has a Kafka cluster available that can also serve as the streaming backbone for your data pipelines.

The streaming architecture follows a three-layer pattern:

  1. Kafka acts as the durable, distributed message bus. Data producers (applications, CDC connectors, IoT devices) publish events to Kafka topics.
  2. Spark Structured Streaming ou Apache Flink runs as a long-lived ilum job that continuously reads from Kafka topics, transforms the data, and writes results downstream.
  3. Iceberg or Delta Lake serves as the sink table format, providing ACID transactions, schema evolution, and time-travel queries on top of object storage (S3, MinIO, HDFS).

This combination delivers low-latency, exactly-once data ingestion pipelines that are fully managed through the ilum platform.

Conditions préalables​

Before building a streaming pipeline, ensure the following components are in place:

  • An ilum cluster with Kafka enabled (this is the default configuration).
  • Iceberg or Delta Lake table format configured in your catalog. You can verify table availability through the Visionneuse SQL.
  • Un Kafka topic populated with data (or a producer actively writing to one).
  • Familiarity with submitting jobs through ilum. See Run a Simple Spark Job for an introduction.

Connecting Spark Jobs to Kafka​

Spark Structured Streaming reads from Kafka using the Kafka format. You must include the Kafka connector package when submitting your job.

Required Spark Package​

Add the following Maven coordinate to your job's Spark Packages configuration in the ilum UI:

org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
pourboire

When submitting through the ilum UI, add this under the Ressources tab in the Spark Packages field. Ilum will resolve and distribute the dependency automatically.

Reading from Kafka​

The core pattern for connecting to a Kafka topic uses readStream with the Kafka format.

PySpark:

Df = (
    étincelle.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "ilum-kafka:9092")
    .option("subscribe", "events")
    .option("startingOffsets", "earliest")
    .load()
)

# Kafka messages are binary key/value pairs. Cast to string for processing.
parsed = Df.selectExpr(
    "CAST(key AS STRING) AS event_key",
    "CAST(value AS STRING) AS event_value",
    "topic",
    "partition",
    « décalage »,
    "timestamp"
)

Scala:

val Df = étincelle.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "ilum-kafka:9092")
  .option("subscribe", "events")
  .option("startingOffsets", "earliest")
  .load()

val parsed = Df.selectExpr(
  "CAST(key AS STRING) AS event_key",
  "CAST(value AS STRING) AS event_value",
  "topic",
  "partition",
  « décalage »,
  "timestamp"
)
note

The default Kafka bootstrap server within an ilum cluster is ilum-kafka:9092. If your Kafka deployment uses a different service name or port, adjust accordingly.

Authentication and Security​

For Kafka clusters configured with SASL or TLS, add the corresponding properties to your Spark configuration:

Df = (
    étincelle.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "ilum-kafka:9093")
    .option("subscribe", "events")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.mechanism", "SCRAM-SHA-512")
    .option("kafka.sasl.jaas.config",
        'org.apache.kafka.common.security.scram.ScramLoginModule required '
        'username="user" password="password";')
    .option("kafka.ssl.truststore.location", "/opt/spark/truststore.jks")
    .option("kafka.ssl.truststore.password", "changeit")
    .load()
)
avertissement

Avoid hardcoding credentials in job code. Use Kubernetes secrets mounted as files or environment variables, and reference them in your Spark configuration.

Exactly-Once Semantics​

Structured Streaming achieves exactly-once processing guarantees through checkpointing et idempotent writes.

Checkpointing Configuration​

Checkpoints track the processing state, including Kafka offsets, so that a restarted job resumes from where it left off without data loss or duplication. Store checkpoints on durable object storage.

requête = (
    parsed.writeStream
    .format("iceberg")
    .outputMode("append")
    .option("checkpointLocation", "s3a://my-bucket/checkpoints/events-pipeline")
    .toTable("spark_catalog.default.events")
)
avertissement

Each streaming query must have a unique checkpoint location. Reusing a checkpoint path across different queries will cause data corruption.

Idempotent Writes to Iceberg​

Iceberg tables support atomic commits, which means each micro-batch is written as a single transaction. If a batch fails partway through, the partial write is rolled back. Combined with checkpointing, this provides end-to-end exactly-once delivery.

Failure Recovery​

When a streaming job fails and is restarted (manually or via ilum's Max Retries setting):

  1. Spark reads the latest checkpoint to determine the last successfully committed Kafka offset.
  2. Processing resumes from the next offset, skipping already-committed data.
  3. No manual intervention or offset management is required.
pourboire

Poser Max Retries on your ilum job to a reasonable value (e.g., 3-5) so that transient failures are automatically recovered without operator intervention.

CDC Patterns​

Change Data Capture (CDC) pipelines replicate changes from operational databases (PostgreSQL, MySQL, MongoDB) into your lakehouse. The typical architecture is:

Source DB --> Debezium --> Kafka --> Structured Streaming --> Iceberg/Delta

Debezium CDC Source​

Debezium is an open-source CDC platform that captures row-level changes from databases and publishes them to Kafka topics. A typical Debezium deployment runs as a Kafka Connect connector alongside your ilum Kafka cluster.

Key Debezium configuration properties:

connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=postgres-host
database.port=5432
database.user=debezium
database.password=secret
database.dbname=mydb
database.server.name=mydb
table.include.list=public.orders,public.customers
topic.prefix=cdc

This produces Kafka topics like cdc.public.orders et cdc.public.customers, each containing insert, update, and delete events.

Upsert with MERGE INTO​

CDC events include inserts, updates, and deletes. To apply these changes to an Iceberg or Delta table, use the foreachBatch sink with a MERGE INTO statement.

PySpark example:

Def upsert_to_iceberg(batch_df, batch_id):
    # Register the micro-batch as a temporary view
    batch_df.createOrReplaceTempView("cdc_batch")

    batch_df.sparkSession.SQL("""
        MERGE INTO spark_catalog.default.orders AS target
        USING cdc_batch AS source
        ON target.id = source.id
        WHEN MATCHED AND source.op = 'd' THEN DELETE
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED AND source.op != 'd' THEN INSERT *
    """)

requête = (
    cdc_stream.writeStream
    .foreachBatch(upsert_to_iceberg)
    .option("checkpointLocation", "s3a://my-bucket/checkpoints/orders-cdc")
    .start()
)
note

Le op field in Debezium events indicates the operation type: c (create), u (update), d (delete), and r (read/snapshot). Adjust the MERGE logic based on your Debezium envelope format.

Example: End-to-End Streaming Pipeline​

The following complete example reads JSON events from a Kafka topic, parses them, and writes them to an Iceberg table.

Complete PySpark Code​

De Pyspark.SQL importation SparkSession
De Pyspark.SQL.functions importation from_json, col
De Pyspark.SQL.types importation StructType, StructField, StringType, DoubleType, TimestampType

étincelle = SparkSession.constructeur\
    .nomappli("streaming-ingestion-pipeline") \
    .getOrCreate()

# Define the expected JSON schema
schéma = StructType([
    StructField("event_id", StringType(), Vrai),
    StructField("user_id", StringType(), Vrai),
    StructField("action", StringType(), Vrai),
    StructField("amount", DoubleType(), Vrai),
    StructField("event_time", TimestampType(), Vrai),
])

# Read from Kafka
raw_stream = (
    étincelle.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "ilum-kafka:9092")
    .option("subscribe", "user-events")
    .option("startingOffsets", "earliest")
    .load()
)

# Parse JSON values
parsed_stream = (
    raw_stream
    .selectExpr("CAST(value AS STRING) AS json_str")
    .select(from_json(col("json_str"), schéma).alias("data"))
    .select("data.*")
    .filtre(col("event_id").isNotNull())
)

# Write to Iceberg table
requête = (
    parsed_stream.writeStream
    .format("iceberg")
    .outputMode("append")
    .option("checkpointLocation", "s3a://my-bucket/checkpoints/user-events")
    .toTable("spark_catalog.default.user_events")
)

requête.awaitTermination()

Submitting as an Ilum Job​

  1. Save the script above as streaming_pipeline.py.
  2. In the ilum UI, create a new job with Job Type se mettre à Spark Job et Langue se mettre à Python.
  3. Upload streaming_pipeline.py under the Ressources onglet.
  4. Add the Kafka connector package under Spark Packages: 
    org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
  5. Under the Configuration tab, add the following Spark properties for Iceberg support: 
    spark.sql.catalog.spark_catalog = org.apache.iceberg.spark.SparkCatalog
    spark.sql.catalog.spark_catalog.type = hive
  6. Submit the job. For detailed submission steps, see Run a Simple Spark Job.

Monitoring Streaming Jobs​

Streaming jobs run continuously, so monitoring is essential:

  • Interface utilisateur Ilum: The job will appear with a RUNNING status. Use the Journaux tab to monitor micro-batch progress and throughput.
  • Interface utilisateur Spark: Access the Spark UI through ilum to view the Structured Streaming tab, which shows input rate, processing rate, and batch duration metrics.
  • Max Retries: Configure automatic restart on failure to maintain pipeline uptime.
pourboire

For long-running streaming jobs, allocate sufficient driver and executor memory to handle state accumulation. Monitor memory usage through the ilum UI and adjust resources as needed.

Apache Flink is supported as a streaming engine in ilum, complementing Spark Structured Streaming. Flink provides true event-at-a-time processing with lower latency and is well suited for complex event processing (CEP), session windowing, and use cases where sub-second response times are critical.

Ilum deploys Flink jobs on Kubernetes using the Flink Kubernetes Operator. Flink applications are submitted through the ilum UI or REST API just like Spark jobs. Ilum handles pod lifecycle management, checkpoint storage configuration, and integration with the shared Kafka cluster and catalog layer.

Key architectural points:

  • Job Manager et Task Manager pods are orchestrated by Kubernetes, with resource allocation managed through ilum's cluster configuration.
  • Flink jobs connect to the same Kafka cluster used by ilum internally (ilum-kafka:9092 by default).
  • Flink integrates with the Ruche Metastore et Iceberg catalog, allowing Flink and Spark to read and write the same tables.
  • Checkpoints and savepoints are stored on the ilum default storage (S3/MinIO/HDFS), enabling job recovery and upgrades without data loss.

Flink SQL provides a declarative way to build streaming pipelines. The following example reads JSON events from a Kafka topic and writes them to an Iceberg table.

-- Register the Kafka source table
CRÉER TABLE kafka_events (
    event_id    STRING,
    user_id     STRING,
    action      CORDE,
    quantité      DOUBLE,
    event_time  HORODATAGE(3),
    WATERMARK FOR event_time COMME event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user-events',
    'properties.bootstrap.servers' = 'ilum-kafka:9092',
    'properties.group.id' = 'flink-ingestion',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false'
);

-- Register the Iceberg sink table
CRÉER TABLE iceberg_events (
    event_id    STRING,
    user_id     STRING,
    action      CORDE,
    quantité      DOUBLE,
    event_time  HORODATAGE(3)
) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'hive_catalog',
    'catalog-type' = 'hive',
    'uri' = 'thrift://ilum-hive-metastore:9083',
    'warehouse' = 's3a://ilum-data/warehouse'
);

-- Continuous streaming insert
INSÉRER DANS iceberg_events
CHOISIR event_id, user_id, action, quantité, event_time
DE kafka_events
OÙ event_id EST NON ZÉRO;
pourboire

Flink SQL jobs can be submitted as SQL scripts through the ilum UI. Set the Job Type À Courageux and provide the SQL statements in the code editor.

For more complex processing logic, use the Flink DataStream API. The following Java example reads from Kafka, applies a tumbling window aggregation, and writes the results to an Iceberg table.

importation org.apache.flink.API.common.eventtime.WatermarkStrategy;
importation org.apache.flink.connector.Kafka.source.KafkaSource;
importation org.apache.flink.connector.Kafka.source.enumerator.initializer.OffsetsInitializer;
importation org.apache.flink.streaming.API.environment.StreamExecutionEnvironment;
importation org.apache.flink.streaming.API.windowing.assigners.TumblingEventTimeWindows;
importation org.apache.flink.streaming.API.windowing.time.Time;

StreamExecutionEnvironment Env = StreamExecutionEnvironment.getExecutionEnvironment();
Env.enableCheckpointing(60000); // checkpoint every 60 seconds

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("ilum-kafka:9092")
    .setTopics("user-events")
    .setGroupId("flink-datastream")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

Env.fromSource(source, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)), "Kafka Source")
    .carte(json -> parseEvent(json))          // custom parsing logic
    .keyBy(event -> event.getUserId())
    .window(TumblingEventTimeWindows.de(Time.minutes(5)))
    .sum("amount")
    .sinkTo(icebergSink);                   // Iceberg sink connector

Env.exécuter("flink-aggregation-pipeline");

Flink provides a native CDC connector that captures database changes directly without requiring a separate Debezium deployment. This simplifies the CDC pipeline from a three-component architecture to two.

Source DB --> Flink CDC Connector --> Iceberg/Delta

Flink SQL example with PostgreSQL CDC:

-- Register a CDC source directly from PostgreSQL
CRÉER TABLE orders_cdc (
    id          INT,
    customer_id INT,
    product     STRING,
    quantité      DOUBLE,
    updated_at  HORODATAGE(3),
    PRIMARY KEY (id) NON ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = 'postgres-host',
    'port' = '5432',
    'username' = 'cdc_user',
    'password' = 'secret',
    'database-name' = 'mydb',
    'schema-name' = 'public',
    'table-name' = 'orders',
    'slot.name' = 'flink_orders_slot'
);

-- Write changes directly to Iceberg with upsert semantics
INSÉRER DANS iceberg_orders
CHOISIR id, customer_id, product, quantité, updated_at
DE orders_cdc;
note

Flink CDC connectors support PostgreSQL, MySQL, MongoDB, Oracle, and SQL Server. Each connector requires the corresponding Flink CDC library to be included in the job dependencies.

Flink achieves exactly-once semantics through its distributed snapshot (Chandy-Lamport) checkpointing algorithm:

  1. Checkpoint barriers flow through the data stream, ensuring consistent state snapshots across all operators.
  2. State backends (RocksDB or heap-based) persist operator state to durable storage (S3/MinIO) on each checkpoint.
  3. Two-phase commit sinks (Kafka, Iceberg) ensure that output records are committed atomically with the checkpoint.

Configure checkpointing in your Flink job:

Env.enableCheckpointing(60000);                          // checkpoint interval (ms)
Env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
Env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
Env.getCheckpointConfig().setCheckpointStorage("s3://ilum-data/flink-checkpoints");
pourboire

For Flink SQL jobs, set checkpointing via Flink configuration properties:

execution.checkpointing.interval=60s
execution.checkpointing.mode=EXACTLY_ONCE
state.checkpoints.dir=s3://ilum-data/flink-checkpoints

Both engines are production-ready for streaming workloads on ilum. Choose based on your requirements:

AspectSpark Structured StreamingApache Flink
Processing ModelMicro-batch (default) or continuousTrue event-at-a-time
LatencySeconds (micro-batch)Milliseconds
State ManagementLimited by micro-batch boundariesFine-grained, key-partitioned state
WindowingTumbling, sliding, sessionTumbling, sliding, session, custom
Complex Event ProcessingBasiqueAdvanced (CEP library)
CDCVia Debezium + KafkaNative CDC connectors
Batch + Stream UnificationStrong (same DataFrame API)Strong (same DataStream/Table API)
Catalog IntegrationHive, Nessie, Unity, Iceberg, Delta, HudiHive, Iceberg
Best ForETL pipelines, unified batch/stream, existing Spark codebasesLow-latency event processing, CEP, native CDC

General guidance:

  • Utiliser Spark Structured Streaming when you have existing Spark batch jobs and want a unified batch/streaming codebase, or when micro-batch latency (seconds) is acceptable.
  • Utiliser Courageux when you need sub-second latency, complex event processing patterns, native CDC without Debezium, or fine-grained state management with event-time processing.
Last updated on