The Table API and Flink SQL

Analogy categoryAnalogies

Why a Relational Layer on Top of Streams

The Table API and Flink SQL are Flink's declarative APIs, built on the same runtime as the DataStream API but letting you describe transformations as relational queries instead of operator chains. A query like SELECT user_id, COUNT(*) FROM clicks GROUP BY user_id compiles down to the same keyed state and operator graph you would hand-write with keyBy and process, but the optimizer picks join strategies, pushes down filters, and manages state layout for you.

🏏

Cricket analogy: It's like a captain giving a bowling change instruction in plain language ('bring on the spinner for the middle overs') rather than micromanaging exact field placements — the team (optimizer) works out the precise mechanics of executing that intent.

Defining Tables and Running Queries

You create tables either from a DataStream via StreamTableEnvironment.fromDataStream, or directly against a connector using a CREATE TABLE DDL statement that specifies a schema, a connector (kafka, filesystem, jdbc), and format options like json or avro. Once registered, tables can be queried with the fluent Table API (table.groupBy($("user_id")).select(...)) or with embedded SQL strings via tableEnv.sqlQuery(...), and both compile to the identical logical plan, so mixing them in one job is common and safe.

🏏

Cricket analogy: Registering a table is like registering a player with the scoring system before a match — once the schema (name, role, batting order) is set, both the scorer's manual tally and the electronic scoreboard (Table API vs SQL) read from the same source of truth.

sql

CREATE TABLE clicks (
  user_id STRING,
  url STRING,
  click_time TIMESTAMP(3),
  WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'clicks',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json'
);

SELECT
  user_id,
  COUNT(*) AS click_count,
  TUMBLE_END(click_time, INTERVAL '1' MINUTE) AS window_end
FROM clicks
GROUP BY user_id, TUMBLE(click_time, INTERVAL '1' MINUTE);

Dynamic Tables and the Streaming-Batch Duality

Flink SQL treats a stream as a continuously changing table called a dynamic table: an INSERT-only stream (like raw clicks) becomes an append-only table, while an aggregation with a GROUP BY produces a changelog table that emits update/retract rows as counts change. This changelog concept, exposed through the toChangelogStream conversion, is what lets the same SQL query run unmodified over bounded batch data or unbounded streaming data — the difference is only whether the result table stabilizes once or keeps updating forever.

🏏

Cricket analogy: It's like a live scorecard during a T20 chase — the run total isn't a fixed number, it's continuously updated (a changelog), whereas the final printed scorecard after the match is a stable, append-only record of what happened.

The Table API and DataStream API can be freely mixed: convert a stream to a table with fromDataStream, run SQL on it, then convert back with toDataStream for custom logic the relational layer can't express, such as calling an external ML model per record.

Catalogs and Connectors

Catalogs let you persist table metadata (schema, connector config) across job restarts and share definitions across teams; Flink ships an in-memory GenericInMemoryCatalog by default but production deployments typically register a HiveCatalog or JDBC catalog so that CREATE TABLE statements survive cluster restarts and multiple jobs can reference the same logical tables without redefining connector properties each time.

🏏

Cricket analogy: It's like a national cricket board maintaining a central player registry — instead of every franchise re-verifying a player's stats before each match, they all reference the same authoritative record.

Aggregations and joins over unbounded streams without a time bound (like GROUP BY user_id alone) accumulate state forever unless you configure idle state retention (table.exec.state.ttl) — in production this is a common cause of unbounded RocksDB growth.

The Table API and Flink SQL are declarative layers compiled to the same runtime as the DataStream API, so they share performance characteristics.
Tables are registered via CREATE TABLE DDL or fromDataStream, and connectors define how data is read from and written to external systems.
Dynamic tables model streams as continuously changing relations; aggregations produce changelog streams with insert/update/delete semantics.
The same SQL query can run over bounded batch or unbounded streaming data — the runtime mode determines whether results are final or continuously updated.
Catalogs (Hive, JDBC, in-memory) persist table metadata so definitions survive restarts and are shared across jobs.
Mixing Table API/SQL with DataStream via fromDataStream/toDataStream conversions is a common pattern for combining declarative and custom logic.
Unbounded aggregations without state TTL configuration can grow state indefinitely — always set retention for production streaming SQL.