From ksqlDB to Flink SQL Without the Guesswork: Meet ksql2flink

Data Streaming · Apache Flink · Confluent

Many teams that built their real-time pipelines on Confluent ksqlDB are now looking at Apache Flink SQL — whether to move to open-source Flink, Confluent Platform’s Flink (CFK/CMF), or Confluent Cloud for Flink. On paper the two look almost the same: both are SQL over Kafka topics. In practice, the migration stalls on the small details — and those small details are exactly where pipelines silently break.

ksql2flink is an tested command-line tool that does this migration for you. It reads your ksqlDB streams, tables, queries, windows, and joins, and writes out ready-to-run Flink SQL — and, just as importantly, it tells you honestly about the parts a human still needs to decide.

Why this migration is harder than it looks

ksqlDB and Flink SQL differ in the places that matter most for correctness:

• Changelogs. A ksqlDB STREAM is append-only; a TABLE is an upsert. In Flink these are two different connectors (kafka vs upsert-kafka) with different key handling.
• Windows. ksqlDB writes WINDOW TUMBLING (SIZE 1 MINUTE). Flink uses windowing table functions like TUMBLE(TABLE t, DESCRIPTOR(rowtime), INTERVAL '1' MINUTE).
• Event time and watermarks. ksqlDB’s ROWTIME and TIMESTAMP='col' map to Flink computed columns and WATERMARK declarations.
• Keys on the wire. How the key is serialized, and how it’s excluded from the value, has to match exactly or your joins quietly produce wrong results.
• Pull queries and custom UDFs. Some things simply have no direct Flink equivalent.

Translate any of these by hand across dozens of statements and you will make a mistake somewhere. That’s the problem ksql2flink was built to remove.

The core idea: convert what’s safe, flag what isn’t

The tool follows one rule: nothing crashes, and nothing vanishes.

Every ksqlDB object that has a well-defined Flink equivalent is converted automatically. Anything that needs human judgement — a pull query, a custom function, a pinned schema ID — is not guessed at. It lands in a migration report with a concrete next step. You always know exactly what was translated and what still needs your eyes.

A quick example

Here’s a typical ksqlDB pipeline — a stream and a windowed aggregate:

CREATE STREAM orders (order_id BIGINT KEY, customer VARCHAR, amount DOUBLE, ts BIGINT)
  WITH (KAFKA_TOPIC='orders', VALUE_FORMAT='JSON', TIMESTAMP='ts');

CREATE TABLE sales_per_min AS
  SELECT customer, COUNT(*) AS order_count, SUM(amount) AS total
  FROM orders WINDOW TUMBLING (SIZE 1 MINUTE)
  GROUP BY customer;

ksql2flink turns the stream into a Flink CREATE TABLE, complete with the event-time column and watermark that ksqlDB implied:

CREATE TABLE orders (
  order_id BIGINT,
  customer STRING,
  amount DOUBLE,
  ts BIGINT,
  rowtime AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR rowtime AS rowtime - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'value.format' = 'json',
  'key.fields' = 'order_id',
  'key.format' = 'raw',
  'value.fields-include' = 'EXCEPT_KEY',
  'scan.startup.mode' = 'earliest-offset'
);

And it splits the single ksqlDB CREATE TABLE AS into the two statements Flink needs — a sink table plus an INSERT INTO — rewriting the window into a table function and adding the window bounds to the upsert key:

CREATE TABLE sales_per_min (
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  customer STRING,
  order_count BIGINT,
  total DOUBLE,
  PRIMARY KEY (window_start, window_end, customer) NOT ENFORCED
) WITH ('connector' = 'upsert-kafka', 'topic' = 'sales_per_min', ...);

INSERT INTO sales_per_min
SELECT window_start, window_end, customer, COUNT(*) AS order_count, SUM(amount) AS total
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(rowtime), INTERVAL '1' MINUTE)) AS orders
GROUP BY customer, window_start, window_end;

Every one of those decisions — the watermark, the EXCEPT_KEY, the window bounds in the primary key — is a place where a hand migration typically goes wrong.

Getting started in three commands

Install it, then point it at a folder of .ksql files:

pip install ksql2flink

# open-source / self-managed Apache Flink
ksql2flink migrate --from-files ./ksql --target oss --out-dir build/flink

# read the results
cat build/flink/tables.sql build/flink/queries.sql build/flink/MIGRATION_REPORT.md

Pick your target with a single flag:

• --target oss — open-source Apache Flink
• --target cfk — Confluent Platform Flink (also emits Kubernetes/CMF resources)
• --target cc — Confluent Cloud for Flink

You can also migrate straight from a running ksqlDB cluster — and that path is strictly read-only. It only issues LIST/DESCRIBE and GET /info, never a CREATE, DROP, INSERT, or TERMINATE. Your production cluster is never touched.

What it converts (and what it flags)

Converted automatically: streams and tables, CREATE ... AS SELECT (into sink DDL + INSERT), tumbling / hopping / session windows, stream-stream, stream-table and table-table joins, PARTITION BY, EXPLODE, custom types, aggregates, and the JSON, DELIMITED, Avro, JSON Schema and Protobuf formats.

Flagged with an action (never silently wrong): pull queries, CREATE CONNECTOR, custom UDFs (for which it generates Java and PyFlink stubs to fill in), lifecycle statements, and pinned schema IDs.

Each run ends with a MIGRATION_REPORT.md that gives you object counts, a per-object status table, a coverage percentage, and a checklist of manual actions.

Verified against a real cluster

This isn’t a paper exercise. Every mapping rule was executed against a real Confluent Platform 8.x + Flink 1.20 cluster — real Kafka with mTLS, a real Schema Registry, and real running Flink jobs — including a genuine cutover where the ksqlDB queries were terminated and the Flink jobs continued the pipelines correctly, verified record by record.

It supports ksqlDB versions 7.2.x through 8.2.x, detects the server version automatically, and warns (rather than fails) outside that range.

The bottom line

Migrating from ksqlDB to Flink SQL doesn’t have to be a slow, error-prone rewrite. ksql2flink handles the mechanical translation, encodes the tricky semantics as tested rules, and hands you an honest report of everything that still needs a human. You spend your time on the decisions that matter — not on hunting down a mis-serialized key at 2 a.m.

Want help planning a ksqlDB-to-Flink migration for your platform? Get in touch with the Alephys team.

Author:Siva Munaga, Solution Architect (and developer of ksql2flink)