Data Streaming: ksqlDB & Flink SQL for Stream Analytics

Overview

The Kafka ecosystem learned an early lesson from Hadoop: SQL dialects dramatically widen the population of engineers who can build production pipelines. ksqlDB and Flink SQL are the two mainstream answers to “I want to query a stream without writing Java topology code.” Both can produce correct results on unbounded data. The difference is in ownership model, operational surface, and what happens when you need to go off-script.

For a Saudi bank running on IBM Event Streams (Kafka 3.x) or Confluent, the deployment picture is asymmetric: ksqlDB runs inside the Kafka ecosystem with minimal new infrastructure; Flink SQL requires a Flink cluster that needs its own cluster management, scaling strategy, and checkpointing store. That is a legitimate operational cost. But Flink SQL also offers SQL semantics the ksqlDB team openly describes as a future target, not a shipped feature—complex multi-stream joins, MATCH_RECOGNIZE pattern detection, savepoints, and first-class SQL DDL for state management.

ksqlDB vs Flink SQL

Choose the engine once, deliberately. Migrating a production streaming SQL application to a different engine is not a weekend project; the SQL dialects differ in ways that matter, and state must be rebuilt from scratch.

Deployment architecture

The two engines slot into an existing Kafka-based platform differently. ksqlDB extends the Kafka footprint; Flink SQL stands beside it.

Writing streaming queries

Both engines distinguish between a stream (an unbounded append log, backed by a Kafka topic) and a table (the latest value per key, backed by a compacted topic or KTable). Every query is declared against one of these abstractions.

-- Declare an existing Kafka topic as a stream
CREATE STREAM card_auth (
  auth_id     VARCHAR KEY,
  customer_id VARCHAR,
  amount      DECIMAL(15,2),
  mcc         VARCHAR,
  ts          BIGINT
) WITH (
  KAFKA_TOPIC  = 'card-auth',
  VALUE_FORMAT = 'AVRO',
  TIMESTAMP    = 'ts'
);

-- Declare a compacted topic as a table (latest-value-per-key)
CREATE TABLE customer_profile (
  customer_id   VARCHAR PRIMARY KEY,
  risk_segment  VARCHAR,
  daily_limit   DECIMAL(15,2)
) WITH (
  KAFKA_TOPIC  = 'customer-master',
  VALUE_FORMAT = 'AVRO'
);

-- Push query: subscribe to enriched output (useful for dashboards)
SELECT a.auth_id, a.amount, c.risk_segment
FROM  card_auth a
JOIN  customer_profile c ON a.customer_id = c.customer_id
EMIT CHANGES;

Flink SQL uses identical logical abstractions but different DDL syntax and connector notation:

-- Source table backed by a Kafka topic (append stream)
CREATE TABLE card_auth (
  auth_id      STRING,
  customer_id  STRING,
  amount       DECIMAL(15,2),
  mcc          STRING,
  ts           TIMESTAMP(3),
  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
  'connector'                     = 'kafka',
  'topic'                         = 'card-auth',
  'properties.bootstrap.servers'  = 'kafka:9092',
  'format'                        = 'avro-confluent',
  'avro-confluent.schema-registry.url' = 'http://schema-registry:8081'
);

-- Lookup table (upsert Kafka connector for changelog-aware reads)
CREATE TABLE customer_profile (
  customer_id   STRING PRIMARY KEY NOT ENFORCED,
  risk_segment  STRING,
  daily_limit   DECIMAL(15,2)
) WITH (
  'connector'   = 'upsert-kafka',
  'topic'       = 'customer-master',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format'  = 'raw',
  'value.format'= 'avro-confluent',
  'value.avro-confluent.schema-registry.url' = 'http://schema-registry:8081'
);

Stateful aggregations

Streaming SQL aggregations are always stateful. The engine must remember previous rows within the scope of a window or a grouping key. Unlike batch SQL, there is no full table scan — only incremental update as each new event arrives. This has two implications: aggregation results are continuously emitted (a retraction stream) and state size is bounded only by the key cardinality and the window duration.

CREATE TABLE customer_spend_10m AS
SELECT
  customer_id,
  WINDOWSTART              AS window_start,
  WINDOWEND                AS window_end,
  COUNT(*)                  AS tx_count,
  SUM(amount)               AS total_spend,
  MAX(amount)               AS max_single_tx
FROM  card_auth
  WINDOW TUMBLING (SIZE 10 MINUTES)
GROUP BY customer_id
EMIT FINAL;         -- only emit once the window closes, not on every row

Schema management

Schema Registry is non-negotiable in production. It enforces Avro or Protobuf compatibility on every topic, which prevents a schema change in one pipeline from silently breaking all downstream consumers. Both ksqlDB and Flink SQL integrate with the Confluent Schema Registry API.

Two compatibility modes matter for financial data:

• BACKWARD — new readers can read old messages. Safe for adding nullable fields. The default for most transaction topics.
• FULL — new readers can read old messages and old readers can read new messages. Required when you have long-running consumers (like compliance archives) that cannot be redeployed quickly.

Windowing in SQL

Both engines support four window types via SQL functions. Flink SQL adds a fifth (CUMULATE) which is particularly useful for regulatory reporting patterns like “running daily total since midnight, updated every minute.”

Stream-table joins

Stream-table joins are the enrichment pattern: every incoming event is enriched with the current state of a reference dataset (customer profile, sanctions list, merchant category). Both engines support this, but the join semantics differ in an important way:

• ksqlDB — stream-table join uses the table state at the time the stream event arrives. There is no time-travel or delayed lookup. This is correct for most enrichment cases.
• Flink SQL — the FOR SYSTEM_TIME AS OF syntax enables temporal joins: look up the reference table at the event’s own timestamp, not the current state. This matters for audit-grade enrichment where you need to prove “what was the counterparty’s sanction status at the time of this transaction”—not today.

-- sanctions_list must be a Flink temporal table with versioned rows
CREATE TABLE sanctions_list (
  entity_id      STRING PRIMARY KEY NOT ENFORCED,
  sanction_flag  BOOLEAN,
  list_date      TIMESTAMP(3),  -- when this version became effective
  WATERMARK FOR list_date AS list_date - INTERVAL '0' SECOND
) WITH ( /* connector props */ );

INSERT INTO payment_screening_output
SELECT
  p.payment_id,
  p.beneficiary_id,
  p.amount,
  s.sanction_flag,
  s.list_date AS sanction_version_at_event_time
FROM  payments p
LEFT JOIN sanctions_list FOR SYSTEM_TIME AS OF p.ts AS s
  ON  p.beneficiary_id = s.entity_id;

Building a pipeline: step by step

1. Design the topology before writing SQL

   Name the input topics, the derived intermediate topics, and the output topics. Define the key for each. SQL makes the code easy to write, but mis-keyed topics still cause silent join misses. Draw the DAG first.

2. Register schemas in Schema Registry

   Define Avro schemas for all source topics before writing DDL. Use BACKWARD compatibility as the default, upgrading to FULL for any topic with long-lived consumers. Schema Registry check on each CREATE TABLE at deploy time, not at runtime.

3. Declare streams and tables with explicit watermarks

   For Flink SQL, always declare the WATERMARK FOR expression on the event-time column. The watermark controls window closure and late-event handling. For ksqlDB, set TIMESTAMP in the WITH clause to use event time rather than processing time.

4. Write and test queries against a local broker

   Use docker-compose with a single-broker Kafka and ksqlDB/Flink SQL CLI. Inject synthetic events representing normal, edge, and late-arrival cases. Validate output topic contents with kafka-console-consumer before deploying upstream.

5. Deploy as persistent queries (ksqlDB) or Flink jobs

   For ksqlDB, use CREATE TABLE/STREAM AS SELECT statements committed to source control and applied via the ksqlDB REST API or Terraform provider. For Flink, package as a JAR or SQL script run via flink run, versioned alongside the DDL catalog.

6. Configure checkpointing and retention

   For ksqlDB, set ksql.streams.state.dir and ensure changelog topics have retention at least as long as the maximum expected restart window. For Flink, configure incremental checkpointing to S3 with interval 30–60 seconds and retained checkpoint count ≥ 3.

7. Wire lag monitoring before go-live

   Both engines expose consumer group lag metrics. Set alerting thresholds at 2x normal lag. A sustained lag increase without throughput increase signals state restore in progress or a slow query (missing index, cartesian join). Catch it before business notices.

Production concerns

The SQL surface is approachable; the operations are not. Three things that catch teams after initial deployment:

Fit alongside batch warehouses

Streaming SQL and batch SQL are complementary, not competing. The operational pattern that works in regulated banking is a lambda-adjacent architecture: the streaming layer handles operational decisions (real-time fraud scoring, limit enforcement) while the batch layer handles the analytical workloads that regulators care about (end-of-day reconciliation, daily transaction file to SAMA, SOX reports).

The bridge between the two is the data lake. Flink SQL has a first-class PRINT sink and a Hudi/Iceberg/Delta connector that writes streaming data into a table format that batch engines (Spark, Trino, BigQuery) can query. ksqlDB can sink to Kafka Connect, which in turn writes to S3 for batch consumption. Neither replaces the other.

When not to use streaming SQL

Streaming SQL is the wrong choice when:

• The query is interactive and ad-hoc. Analysts running exploratory queries against a ksqlDB server will encounter latency, cost, and resource contention it was not designed for. Use Trino or BigQuery against a data lake for exploratory work.
• The logic requires complex imperative state management. A ksqlDB or Flink SQL query that requires a custom UDF for every meaningful operation is a signal that the Kafka Streams Java API is a better fit. SQL abstracts state; sometimes you need to manipulate it directly.
• You need cross-window joins with large state on both sides. Joining two streams where both have multi-hour windows and millions of distinct keys materialises enormous state on both sides. Unless you have a clear throughput and state budget, this pattern causes production incidents.
• The pipeline is read-once and discarded. A one-time backfill or a quarterly regulatory extract does not justify the operational overhead of a persistent streaming query. Write a Spark job.