Home About Me

Flink SQL in Practice: Dynamic Tables, Continuous Queries, and the Details That Matter

2026-05-07

The notes below are based on Flink 1.15.

Offline vs real-time computation

Flink SQL's biggest advantage is real-time computation. Hive SQL, by contrast, works on offline data. That difference matters because it can simplify real-time analytics inside an organization.

  • Offline: the input table clicks is static. A SQL statement runs once and finishes; the output is fixed.
  • Real-time: the input table receives data continuously. Each new row can trigger the query again, and the output table changes with it. Both input and output are live.

At a high level, the data source is the difference: offline data arrives all at once, real-time data keeps arriving. In streaming mode, output updates as input changes.

Dynamic tables are the central idea behind stream processing in Flink SQL. A table whose data changes is a dynamic table. Without it, Flink SQL could not process streams. You define a table over a real-time data stream; if data keeps being appended or changing, that table is dynamic.

SQL parsing engine

Flink SQL uses Apache Calcite as its parsing engine. Calcite is an open-source SQL parser. It does not handle storage or computation; it parses SQL and helps turn it into executable code. It turns a statement into a tree (AST) and applies operations to that tree, eventually translating the SQL logic into Flink code.

<table> <thead> <tr> <th>Engine</th> <th>Parsing</th> <th>Optimization</th> <th>Used by</th> </tr> </thead> <tbody> <tr> <td>Apache Calcite</td> <td>yes</td> <td>yes</td> <td>Flink SQL, Hive SQL</td> </tr> <tr> <td>ANTLR</td> <td>yes</td> <td>no</td> <td>Spark SQL (Catalyst handles optimization)</td> </tr> </tbody> </table>

Calcite is more complete: it parses and optimizes. ANTLR is just a parser.

Calcite processes SQL in roughly four steps:

<table> <thead> <tr> <th>Step</th> <th>Name</th> <th>What happens</th> </tr> </thead> <tbody> <tr> <td>1</td> <td>Parser</td> <td>Uses Java CC to turn SQL into an unvalidated AST (also called SQL Node). The tree has not been checked yet.</td> </tr> <tr> <td>2</td> <td>Validate</td> <td>Checks whether the AST is legal, for example whether referenced fields and tables exist and functions are written correctly. After validation it produces a RelNode.</td> </tr> <tr> <td>3</td> <td>Optimize</td> <td>Optimizes the logical plan using rule-based optimization (RBO) or cost-based optimization (CBO) to reach a physical plan. This step is optional in theory, but most parsers use it.</td> </tr> <tr> <td>4</td> <td>Execute</td> <td>Converts the physical plan into code that a specific platform can run.</td> </tr> </tbody> </table>

The full Flink SQL execution flow looks like this:

SQL statement
  ↓ SQL Parser (to AST)
  ↓ SQL Validate (check against metadata)
  ↓ Logical Plan
  ↓ Optimized Logical Plan (Calcite rules + Flink-specific rules)
  ↓ Flink Physical Plan
  ↓ Flink Execution Plan (operator code)
  ↓ Execute

DDL statements

Flink SQL DDL mainly covers creating, altering, and dropping tables. In practice, CREATE TABLE is the one you use most.

Dependencies often include:

<table> <thead> <tr> <th>Dependency</th> <th>Purpose</th> </tr> </thead> <tbody> <tr> <td>Flink SQL package</td> <td>Core</td> </tr> <tr> <td>File System Connector</td> <td>Work with HDFS files</td> </tr> <tr> <td>Hadoop-related dependencies</td> <td>HDFS support</td> </tr> <tr> <td>Flink JSON dependency</td> <td>Needed when the data source is JSON</td> </tr> </tbody> </table>

Check the Flink version carefully. Connector options can change between minor versions, and the documentation version must match the dependency version in code.

Create the execution environment with StreamTableEnvironment.create(), and specify batch or streaming mode:

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = StreamTableEnvironment.create(env)

The basic table creation syntax is:

CREATE TABLE table_name (
  column_name column_type,
  ...
) WITH (
  'connector' = 'xxx',
  ...
)

A complete example: read from File System and print output. This combination is common in test environments because it reads from HDFS files and prints results to the console, which makes logic easy to verify.

First create an input table with the file path and format:

CREATE TABLE file_source (
  name STRING,
  age INT
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://namenode:9000/path/to/data.json',
  'format' = 'json',
  'json.fail-on-missing-field' = 'true',
  'json.ignore-parse-errors' = 'false'
)

Then create an output table that prints results:

CREATE TABLE print_sink (
  age INT,
  cnt BIGINT
) WITH (
  'connector' = 'print'
)

Now write the business logic: count how many people there are for each age.

INSERT INTO print_sink
SELECT age, COUNT(*) AS cnt
FROM file_source
GROUP BY age

Common pitfalls:

<table> <thead> <tr> <th>Problem</th> <th>Cause</th> </tr> </thead> <tbody> <tr> <td>Table not found</td> <td>Flink SQL table names are case-sensitive. If you define a lowercase name, use lowercase when referencing it.</td> </tr> <tr> <td>Column type mismatch</td> <td>COUNT returns BIGINT. Do not define the output column as INT.</td> </tr> <tr> <td>HDFS file cannot be read</td> <td>Use the full path.</td> </tr> <tr> <td>Results keep updating</td> <td>In streaming mode, grouped aggregation keeps triggering computation. This is normal.</td> </tr> <tr> <td>Batch and streaming results differ</td> <td>They use different execution logic. Batch mode returns only the final result.</td> </tr> </tbody> </table>

Table types

<table> <thead> <tr> <th>Table type</th> <th>Description</th> <th>Typical scenario</th> </tr> </thead> <tbody> <tr> <td>Static table</td> <td>Data does not change in real time</td> <td>File System table in batch mode</td> </tr> <tr> <td>Dynamic table</td> <td>Data changes in real time</td> <td>Kafka or File System table in streaming mode</td> </tr> <tr> <td>Versioned table</td> <td>Dynamic table with primary key and event time that records history</td> <td>Dual-stream join validation</td> </tr> <tr> <td>Temporal table function</td> <td>Accesses historical data of a dynamic table; only append-only tables are supported</td> <td>Dual-stream join</td> </tr> </tbody> </table>

A static table is generally defined in batch mode. A File System table created in batch mode is static. Its data can still change, for example when new data is added daily or hourly, but that change is offline and predictable, not real-time.

A dynamic table is the opposite: its data changes in real time. Kafka or File System tables created in streaming mode are dynamic. Dynamic tables are the key to stream processing in Flink SQL; without them there is no streaming computation.

Versioned tables were introduced in Flink 1.12. In Flink 1.11 the concept was called a temporal table and was supported only by the Blink engine. A versioned table is essentially a dynamic table with a primary key and event-time attribute that records historical values for each key. It is similar to a Hive history table. With a versioned table, you can query historical data at a point in time. Creating one takes two steps: define the primary key with PRIMARY KEY, then define the event-time field with WATERMARK. Versioned tables are mainly used for dual-stream join validation.

A temporal table function also accesses historical data, but it works on ordinary dynamic tables rather than versioned tables. It can only be used with append-only dynamic tables and cannot be created with SQL DDL; you must register it through the Table API. Its main scenario is also dual-stream joins.

Continuous queries

A continuous query is best understood in contrast to a one-time query.

In offline scenarios, you run SQL against a static table, the computation finishes, and it does not run again. That is a one-time query.

In real-time scenarios, you run SQL against a dynamic table. Whenever new data arrives, computation is triggered again. Data keeps arriving, so computation keeps triggering. This kind of repeatedly executed query is a continuous query.

In short: when dynamic table data changes, the SQL runs again. Both input and output are dynamic tables.

A continuous query over a real-time stream generally has three stages:

<table> <thead> <tr> <th>Step</th> <th>Operation</th> </tr> </thead> <tbody> <tr> <td>1</td> <td>Real-time stream to dynamic input table (define a table over the stream)</td> </tr> <tr> <td>2</td> <td>Dynamic input table to continuous query execution to dynamic output table (the query uses state to store intermediate results)</td> </tr> <tr> <td>3</td> <td>Dynamic output table to data stream output</td> </tr> </tbody> </table>

Case one: ordinary grouped aggregation. The scenario is counting occurrences by name.

When the first batch arrives, tom appears once and the output is tom 1; jack appears once and the output is jack 1. When the second batch arrives, tom appears again. Since tom has already appeared, the previous result is updated to tom 2; jack does not change. Then jessica arrives, so the output is jessica 1.

In this case the output table contains both inserts and updates, because the same person can appear again.

Case two: tumbling window aggregation. The scenario is a one-hour tumbling window with grouped aggregation each hour.

SELECT name,
       TUMBLE_START(event_time, INTERVAL '1' HOUR) AS window_start,
       COUNT(*) AS cnt
FROM clicks
GROUP BY name, TUMBLE(event_time, INTERVAL '1' HOUR)

Because time only moves forward and different time windows do not overlap, all output data is inserts and there are no updates. This is called an append-only stream.

The output of a continuous query can be understood as a change log data stream, similar to MySQL binlog, recording data change operations.

<table> <thead> <tr> <th>Operation</th> <th>Marker</th> <th>Meaning</th> </tr> </thead> <tbody> <tr> <td>Insert</td> <td>+I</td> <td>Insert new data</td> </tr> <tr> <td>Update before</td> <td>-U</td> <td>Retract old data</td> </tr> <tr> <td>Update after</td> <td>+U</td> <td>Write new data</td> </tr> <tr> <td>Delete</td> <td>-D</td> <td>Delete data</td> </tr> </tbody> </table>

Data types

Flink SQL supports three classes of data types: basic types, composite types, and custom types. In practice, basic and composite types are enough; custom types are rarely needed.

Basic types include:

  • String: use STRING.
  • Binary: BINARY, VARBINARY.
  • Numeric:
<table> <thead> <tr> <th>Type</th> <th>Description</th> </tr> </thead> <tbody> <tr> <td>INT / BIGINT</td> <td>Integers, with the same value ranges as in Java</td> </tr> <tr> <td>FLOAT</td> <td>Single-precision floating point</td> </tr> <tr> <td>DOUBLE</td> <td>Double-precision floating point</td> </tr> <tr> <td>DECIMAL(p, s)</td> <td>Financial use cases such as product prices; p is total digits, s is decimal digits, with higher precision</td> </tr> </tbody> </table>
  • Boolean: BOOLEAN
  • Null: NULL
  • Date and time:
<table> <thead> <tr> <th>Type</th> <th>Description</th> <th>Example</th> </tr> </thead> <tbody> <tr> <td>DATE</td> <td>Year, month, day without time zone</td> <td>2026-04-21</td> </tr> <tr> <td>TIME</td> <td>Hour, minute, second without time zone</td> <td>15:30:00</td> </tr> <tr> <td>TIMESTAMP</td> <td>Year, month, day, hour, minute, second without time zone. Seconds default to 6 digits of precision; 3 is often enough.</td> <td>2026-04-21 15:30:00.123</td> </tr> <tr> <td>TIMESTAMP_LTZ</td> <td>Year, month, day, hour, minute, second with local time zone. Recommended for production.</td> <td>2026-04-21 15:30:00.123</td> </tr> <tr> <td>INTERVAL</td> <td>Time interval, used to add or subtract offsets from time</td> <td>-</td> </tr> </tbody> </table>

The difference between TIMESTAMP and TIMESTAMP_LTZ matters. TIMESTAMP has no time zone. If you have not set a time zone, it uses UTC by default, which is 8 hours behind China time (UTC+8). TIMESTAMP_LTZ reads the local time zone. It is better to use TIMESTAMP_LTZ directly.

You can set the time zone manually in code:

tableEnv.getConfig.setLocalTimeZone(TimeZone.getTimeZone("Asia/Shanghai"))

Interval is used to shift time, such as adding or subtracting days, months, or hours:

<table> <thead> <tr> <th>Usage</th> <th>Example</th> </tr> </thead> <tbody> <tr> <td>Add 1 second</td> <td>SELECT order_time + INTERVAL '1' SECOND FROM orders</td> </tr> <tr> <td>Add 1 minute</td> <td>SELECT order_time + INTERVAL '1' MINUTE FROM orders</td> </tr> <tr> <td>Add 1 day 2 hours</td> <td>SELECT order_time + INTERVAL '1 2' DAY TO HOUR FROM orders</td> </tr> <tr> <td>Add 2 hours 30 minutes</td> <td>SELECT order_time + INTERVAL '2:30' HOUR TO MINUTE FROM orders</td> </tr> </tbody> </table>

Composite types:

<table> <thead> <tr> <th>Type</th> <th>Similar to</th> <th>Description</th> </tr> </thead> <tbody> <tr> <td>ARRAY</td> <td>Java array</td> <td>Array</td> </tr> <tr> <td>MAP<K, V></td> <td>Java Map</td> <td>Key-value pairs</td> </tr> <tr> <td>MULTISET</td> <td>Java List</td> <td>Collection that allows duplicate elements</td> </tr> <tr> <td>ROW</td> <td>Java custom object</td> <td>Row object</td> </tr> </tbody> </table>

These are used heavily in Hive, and Flink usage is similar.

Encoding dynamic table output

When a dynamic output table is converted into a data stream, its change behavior must be encoded. Flink SQL supports three encodings, corresponding to three different data streams.

<table> <thead> <tr> <th>Stream</th> <th>Description</th> <th>Insert</th> <th>Update</th> <th>Delete</th> <th>Primary key required</th> </tr> </thead> <tbody> <tr> <td>Append Only</td> <td>Pure append stream</td> <td>yes</td> <td>no</td> <td>no</td> <td>no</td> </tr> <tr> <td>Retract</td> <td>Retract stream</td> <td>yes</td> <td>yes (two records)</td> <td>yes</td> <td>no</td> </tr> <tr> <td>Upsert</td> <td>Insert-update stream</td> <td>yes</td> <td>yes (one record)</td> <td>yes</td> <td>yes</td> </tr> </tbody> </table>

Append Only streams contain only inserts, with no updates or deletes. They fit time-window aggregation or ordinary real-time ETL without grouping.

Retract streams contain inserts, updates, and deletes. Updates are special: two records are sent, first a -U (Update Before) to retract old data, then a +U (Update After) to add new data. Every update is therefore two steps. If a downstream system continues processing this stream, it must handle -U and +U correctly; otherwise data will be wrong. Grouped aggregation produces a Retract stream when the same key can appear multiple times and results need updating.

Upsert streams also contain inserts, updates, and deletes, but an update sends only one record: the new data overwrites the old data without retracting first. This is more efficient than Retract. Upsert requires a primary key on the table.

How to choose:

<table> <thead> <tr> <th>Scenario</th> <th>Recommended stream</th> </tr> </thead> <tbody> <tr> <td>Time-window aggregation without grouping</td> <td>Append Only</td> </tr> <tr> <td>Grouped aggregation with updates</td> <td>Retract or Upsert</td> </tr> <tr> <td>Grouped aggregation with primary key and frequent updates</td> <td>Upsert, for efficiency</td> </tr> </tbody> </table>

Practical notes: if the SQL uses time-window aggregation or no grouping, output is Append Only and a normal connector works. If the SQL uses grouped aggregation, updates are produced and Retract or Upsert is needed. When writing to Kafka, choose the upsert-kafka connector; a normal Kafka connector does not support this. If you connect to your own third-party storage system, you must parse the markers (+I, -U, +U, -D) in the data stream yourself and write to storage as needed.

Date and time functions

Common functions:

<table> <thead> <tr> <th>Function</th> <th>Description</th> </tr> </thead> <tbody> <tr> <td>CURRENT_DATE</td> <td>Current date</td> </tr> <tr> <td>CURRENT_TIME / LOCALTIME</td> <td>Current time (hour, minute, second)</td> </tr> <tr> <td>CURRENT_TIMESTAMP / LOCALTIMESTAMP</td> <td>Current timestamp (year, month, day, hour, minute, second, millisecond, with time zone)</td> </tr> <tr> <td>NOW()</td> <td>Current timestamp, same effect as TIMESTAMP</td> </tr> <tr> <td>CURRENT_ROW_TIMESTAMP()</td> <td>Current row timestamp</td> </tr> <tr> <td>CURRENT_PROCTIME()</td> <td>Processing time, when the program processes this data</td> </tr> </tbody> </table>

These depend on the time zone. With UTC+8 set, they return Beijing time.

If the timestamp has more millisecond digits than you want, format it with DATE_FORMAT:

SELECT DATE_FORMAT(order_time, 'yyyy-MM-dd HH:mm:ss') FROM orders

The difference between CURRENT_TIMESTAMP and CURRENT_ROW_TIMESTAMP:

<table> <thead> <tr> <th>Scenario</th> <th>CURRENT_TIMESTAMP</th> <th>CURRENT_ROW_TIMESTAMP</th> </tr> </thead> <tbody> <tr> <td>Streaming mode</td> <td>Each row gets its own timestamp</td> <td>Each row gets its own timestamp (same)</td> </tr> <tr> <td>Batch mode</td> <td>Time is taken once when the query starts and used for the whole batch</td> <td>Each row gets its own timestamp</td> </tr> </tbody> </table>

Set the time zone manually:

tableEnv.getConfig.setLocalTimeZone(TimeZone.getTimeZone("Asia/Shanghai"))

If you do not set it, the program reads the machine time zone and may get UTC, which is 8 hours behind Beijing and easy to turn into a bug.

Connectors and test data

Common connectors:

<table> <thead> <tr> <th>Connector</th> <th>Purpose</th> </tr> </thead> <tbody> <tr> <td>filesystem</td> <td>Read and write local or HDFS files</td> </tr> <tr> <td>kafka</td> <td>Kafka source or destination</td> </tr> <tr> <td>upsert-kafka</td> <td>Kafka with Upsert support</td> </tr> <tr> <td>print</td> <td>Print to console, for testing</td> </tr> <tr> <td>datagen</td> <td>Generate test data automatically</td> </tr> </tbody> </table>

datagen is useful for running streaming tests locally. To test update logic in grouped aggregation, you need a field that produces duplicate data. You can limit its value range:

CREATE TABLE orders (
  order_id BIGINT,
  price DECIMAL(10, 2),
  order_time TIMESTAMP
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'order_id.gen-kind' = 'sequence',
  'order_id.gen.min' = '100',
  'order_id.gen.max' = '105'
)

This makes order_id cycle between 100 and 105, which is handy for testing update scenarios triggered by duplicate keys.

Common errors:

<table> <thead> <tr> <th>Error</th> <th>Fix</th> </tr> </thead> <tbody> <tr> <td>Column type mismatch</td> <td>COUNT returns BIGINT, so do not define the column as INT.</td> </tr> <tr> <td>Table not found</td> <td>Usually a case mismatch, or the table has not been created.</td> </tr> <tr> <td>Time zone is 8 hours off</td> <td>Set Asia/Shanghai.</td> </tr> <tr> <td>Upsert primary key cannot be null</td> <td>When using Upsert streams, add a NOT NULL constraint to the primary key field.</td> </tr> </tbody> </table>

Related Posts