Stream processing

Applies to VAST Cluster 5.5.

Apache Flink and Spark Structured Streaming read and write Kafka topics through their Kafka connectors, so they can use VAST Event Broker topics as streaming sources and sinks. Both connectors embed the Apache Kafka Java client, which turns idempotent producing on by default, and both offer exactly-once modes built on Kafka transactions. Event Broker supports neither, so the sinks need explicit settings. This page gives the settings for each engine, what won't work, and a VAST alternative: reading the topic as a table instead of as a stream.

The examples use Apache Flink 2.2 (flink-sql-connector-kafka 5.0.0-2.2, Flink SQL) and Spark Structured Streaming 3.5.1 (spark-sql-kafka-0-10_2.13 3.5.1), unencrypted: reading from and writing to Event Broker topics, committed offsets, restarting from a checkpoint, and the exactly-once and default-idempotence failure modes. The PyFlink DataStream job on this page, secured connections and Kafka Streams aren't covered.

Requirements

Both sinks are at-least-once at best: after a failure and restart, the output topic can hold duplicates. Make the readers of the output tolerate them, for example with a unique key you deduplicate on.

The connectors' consumers need nothing special. Every record a consumer gets from Event Broker is committed, whatever isolation level it asks for. See Consuming.

Apache Flink

Use the Flink Kafka connector (flink-connector-kafka) that matches your Flink version. KafkaSource reads Event Broker topics like any Kafka topics. On KafkaSink, set the delivery guarantee and pass the producer settings as properties:

import os

from pyflink.common import Types, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.base import DeliveryGuarantee
from pyflink.datastream.connectors.kafka import (
    KafkaOffsetsInitializer, KafkaRecordSerializationSchema, KafkaSink, KafkaSource)

BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]

env = StreamExecutionEnvironment.get_execution_environment()
env.enable_checkpointing(10_000)   # AT_LEAST_ONCE needs checkpointing

source = (KafkaSource.builder()
          .set_bootstrap_servers(BOOTSTRAP)
          .set_topics("docs-stream-in")
          .set_group_id("docs-stream-flink")
          .set_starting_offsets(KafkaOffsetsInitializer.earliest())
          .set_value_only_deserializer(SimpleStringSchema())
          .build())

sink = (KafkaSink.builder()
        .set_bootstrap_servers(BOOTSTRAP)
        .set_record_serializer(KafkaRecordSerializationSchema.builder()
                               .set_topic("docs-stream-out")
                               .set_value_serialization_schema(SimpleStringSchema())
                               .build())
        .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .set_property("enable.idempotence", "false")
        .set_property("compression.type", "none")
        .build())

stream = env.from_source(source, WatermarkStrategy.no_watermarks(), "events")
stream.map(lambda value: value.upper(), output_type=Types.STRING()).sink_to(sink)
env.execute("docs-stream-flink")

In Java, the same builder methods are setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) and setProperty("enable.idempotence", "false").

In Flink SQL, Kafka client settings take the properties. prefix:

CREATE TABLE events_out (
  `order_id` BIGINT,
  `status` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'docs-stream-out',
  'properties.bootstrap.servers' = '<vip-1>:9092,<vip-2>:9092',
  'properties.enable.idempotence' = 'false',
  'properties.compression.type' = 'none',
  'sink.delivery-guarantee' = 'at-least-once',
  'format' = 'json'
);

Things to know:

• The KafkaSink default guarantee is NONE; the SQL connector's default is at-least-once. Neither uses transactions. The Flink SQL job on this page copies all 1,000 events from one topic to another with at-least-once, and commits the source offsets on each checkpoint.
• exactly-once needs transactions, so it can't work, and it fails in a confusing way:
  • With 'properties.enable.idempotence' = 'false', the Kafka client refuses the configuration (ConfigException: Cannot set a transactional.id without also enabling idempotence), and the job restarts over and over.
  • Without that line, the job shows RUNNING but writes nothing and commits no offsets: its producer waits for a transaction coordinator that Event Broker never provides, and logs no error.
• With commit.offsets.on.checkpoint (on by default), the source commits its offsets to the consumer group on each checkpoint, which Event Broker supports. Flink doesn't rely on these offsets for fault tolerance: it restores from its own checkpoints. Each group counts toward the 256 consumer groups a Kafka view allows (Limits).
• Partition discovery (partition.discovery.interval.ms) does no harm, but an Event Broker topic never gains partitions.
• For SASL/PLAIN or TLS, pass the Kafka client security settings as properties on both the source and the sink (properties. prefixed in SQL). See Security.

Spark Structured Streaming

The Spark Kafka source and sink (spark-sql-kafka-0-10) take Kafka client settings as options with the kafka. prefix. Set the producer settings on the sink:

import os

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("docs-stream-spark").getOrCreate()
BOOTSTRAP = os.environ["KAFKA_BOOTSTRAP"]

events = (spark.readStream.format("kafka")
          .option("kafka.bootstrap.servers", BOOTSTRAP)
          .option("subscribe", "docs-stream-in")
          .option("startingOffsets", "earliest")
          .load())

query = (events.selectExpr("key", "upper(CAST(value AS STRING)) AS value")
         .writeStream.format("kafka")
         .option("kafka.bootstrap.servers", BOOTSTRAP)
         .option("kafka.enable.idempotence", "false")
         .option("kafka.compression.type", "none")
         .option("topic", "docs-stream-out")
         .option("checkpointLocation", "/tmp/docs-stream-spark")
         .start())
query.awaitTermination()

Things to know:

• The Spark Kafka sink is at-least-once by design, so there's no exactly-once mode to turn off.
• With kafka.enable.idempotence set to false, the job on this page writes every event to the output topic, and a second run picks up from its checkpoint. Without that option, a write fails within seconds with UnsupportedVersionException: The broker does not support INIT_PRODUCER_ID.
• New events can take up to about a minute to be seen. Spark plans each micro-batch, and each trigger(availableNow=True) run, from the end offsets the broker reports, and on Event Broker those can trail the newest events by tens of seconds (10 to 55 seconds after a burst of writes), even though a consumer can already read them. A run that starts right after a burst of events may process none of them; the next batch picks them up. See consuming.
• Spark tracks its position in its checkpoint location, not in a Kafka consumer group.
• startingOffsets accepts earliest, latest or per-partition offsets. Don't use startingOffsetsByTimestamp or startingTimestamp.
• For SASL/PLAIN or TLS, add the client settings with the kafka. prefix, for example kafka.security.protocol and kafka.sasl.mechanism, on both the source and the sink.

Read the topic table instead

Every Event Broker topic is also a VAST DataBase table. If your job reads a topic to load it into a table, or to run batch analytics on it, you may not need a stream at all: Spark and Trino can read the topic table directly through the VAST DataBase connectors, with filters pushed down to the cluster. A query sees every committed event so far. See Query topics as tables, Spark and Trino.

To get typed columns from JSON events without a parse job, use blob expansion.

A stream job is still the right tool when you need per-event processing with low latency, or when the output must be another topic.

Kafka Streams and ksqlDB

There's no worked example for Kafka Streams, or for ksqlDB, which is built on it. A Kafka Streams application creates its own internal topics through the Admin API: compacted changelog topics for its state stores, and repartition topics. It purges the repartition topics with DeleteRecords, which Event Broker doesn't offer, so they would keep their data until retention removes it, and Streams creates them without a time limit by default. Its producers need producer.enable.idempotence=false and producer.compression.type=none. Use processing.guarantee=at_least_once (the default): exactly_once_v2 needs transactions.

What won't work

• Flink EXACTLY_ONCE, Flink SQL exactly-once and Kafka Streams exactly_once_v2: they need transactions.
• Sinks with the client's default producer settings: idempotence is on by default. Spark fails at once with The broker does not support INIT_PRODUCER_ID.

See also

• Integrations: the checklist for any Kafka tool.
• Producing and Consuming
• Kafka compatibility
• Query topics as tables
• Apache Flink Kafka DataStream connector and Kafka SQL connector
• Spark Structured Streaming + Kafka integration guide