Learn Apache Flink - Table API & SQL

#Introduction

The last four episodes dealt with the imperative DataStream API. Episode 9 offers a much shorter route: Flink SQL. With SQL, you write streaming queries — complete with windowing, joins, and aggregations — without writing a single line of manual operator logic. For analytics, this is usually the only API you need.

We'll get to know TableEnvironment and catalogs, write DDL to declare sources and sinks, then apply windowing, joins, and aggregations. At the end, we'll touch on temporal tables and CDC patterns — how Flink sees a constantly changing database table as a stream.

#TableEnvironment and Catalogs

#The Environment for Executables

TableEnvironment is the entry point for all queries. There are two modes: streaming (the default for Flink) and batch:

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
 
EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);
 
tableEnv.executeSql("CREATE TABLE orders (user_id STRING, amount BIGINT, event_ts TIMESTAMP(3))");
tableEnv.executeSql("INSERT INTO agg_orders SELECT user_id, SUM(amount) FROM orders ...");

tableEnv.executeSql runs DDL, DML, and query statements alike. This is the main Table API program pattern: declare, execute, done.

#Catalogs for Centralized Metadata

A catalog stores table metadata so queries can be written without repeating DDL in every job. Flink has built-in catalogs (for example for Hive) and generic JDBC catalogs. With catalogs, teams share the same table definitions — a single source of truth.

CREATE CATALOG orders_catalog WITH (
  'type' = 'generic_in_memory',
  'default-database' = 'analytics'
);
USE CATALOG orders_catalog;

#DDL and DML with Flink SQL

#CREATE TABLE for Sources and Sinks

DDL in Flink SQL combines the data schema and connector configuration in a single statement:

CREATE TABLE orders (
  user_id  STRING,
  amount   BIGINT,
  event_ts TIMESTAMP(3),
  WATERMARK FOR event_ts AS event_ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

Note the WATERMARK FOR event_ts clause — the watermark concept from episode 5 expressed directly in DDL. Without this line, event time windowing won't work.

#INSERT INTO for Writing Results

Query results are written to a sink table with INSERT:

INSERT INTO agg_orders
SELECT user_id, SUM(amount) AS total
FROM orders
GROUP BY user_id, TUMBLE(event_ts, INTERVAL '5' MINUTE);

#Streaming SQL Patterns

#Windowing with Built-in Functions

Flink SQL provides the TUMBLE, HOP, and SESSION window functions that are used directly in GROUP BY:

SELECT user_id, COUNT(*) AS jumlah, SUM(amount) AS total
FROM orders
GROUP BY user_id, TUMBLE(event_ts, INTERVAL '1' MINUTE);
 
SELECT user_id, COUNT(*) AS aktivitas
FROM clicks
GROUP BY user_id, SESSION(event_ts, INTERVAL '30' MINUTE);

TUMBLE(event_ts, INTERVAL '1' MINUTE) is a one-minute tumbling window, and SESSION groups events by activity gap. This windowing syntax is far more concise than the DataStream API.

#Joins Between Streams

SQL also simplifies joins between streams:

SELECT o.order_id, o.amount, p.status
FROM orders o
JOIN payments p
  ON o.order_id = p.order_id
  AND o.event_ts BETWEEN p.event_ts - INTERVAL '5' MINUTE
  AND p.event_ts + INTERVAL '5' MINUTE;

The interval join pattern above limits matches to a certain time window — important because streams don't have clear boundaries.

#Temporal Tables and CDC

A temporal table is a table whose contents change over time — ideal for dimension data. CDC (Change Data Capture) lets Flink read database changes through Debezium:

CREATE TABLE users (
  id   INT PRIMARY KEY NOT ENFORCED,
  nama STRING,
  updated_at TIMESTAMP(3)
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'flink',
  'password' = 'secret',
  'database-name' = 'app',
  'table-name' = 'users'
);

The mysql-cdc connector turns every row change in MySQL into an event stream — a "living" table that keeps updating without manual polling. This is the foundation of modern change-data-capture architectures.

#Using the SQL Client

For quick experiments without writing Java code, use Flink's built-in SQL client:

$FLINK_HOME/bin/sql-client.sh

Inside the SQL client prompt, you can write DDL and queries interactively. The sql-client.sh command is the fastest tool for validating queries before putting them into an application.

#Conclusion

Episode 9 introduced Flink's declarative path: TableEnvironment and catalogs for metadata, DDL combining schemas and connectors, the TUMBLE, HOP, and SESSION window functions, joins between streams, and temporal tables and CDC for changing data. Flink SQL lets streaming analytics be written in a few lines.

The key takeaways:

• TableEnvironment is the entry point for all Table API and SQL programs.
• Catalogs store table definitions so they can be reused across jobs.
• Watermarks are declared in DDL with the WATERMARK FOR clause.
• TUMBLE, HOP, and SESSION express windowing directly in SQL.
• The mysql-cdc connector turns database changes into an event stream.

In the next episode, episode 10, we'll discuss complex event processing (CEP) — defining patterns with the Flink CEP library, matching sequential events, applying them to fraud detection and anomaly detection, and handling timed patterns and pattern states. This is where Flink becomes the "brain" that detects hidden patterns in the data stream.