Definition
- CREATE TABLE – Declares column names and SQL types, optional computed columns, metadata columns, and a primary key.
- WITH clause – A map of string properties binding the logical table to an external system via connector and format settings.
- Connector – Selects the external system:
kafka,jdbc,filesystem,upsert-kafka,elasticsearch, and others. - Format – Selects serialization:
json,avro,csv, ordebezium-jsonfor CDC sources. - WATERMARK – Declares the event-time column and how far behind the watermark trails, enabling event-time windows and temporal joins.
Declaring a Flink SQL table is like filling out a form at the National Bank of India that tells the system three things about a data pipe. The schema is the list of fields each record has. The WITH clause is the plumbing: the connector says which external pipe this is, and the format says what language the messages are written in. The watermark line tells the system which field is the real clock and how many seconds of lateness to tolerate.
Understanding the Concepts
What does CREATE TABLE look like?
The CREATE TABLE statement starts with column definitions, giving each column a name and a SQL type. The WITH clause binds it to an external system.
CREATE TABLE transactions (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
txn_type STRING,
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-transactions',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'json',
'scan.startup.mode' = 'latest-offset'
);
What are the connector options?
The connector property selects the external system. Each connector has its own required properties.
-- Kafka connector
CREATE TABLE kafka_transactions (
txn_id STRING, account_id STRING, amount DECIMAL(15, 2),
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-transactions',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'json',
'scan.startup.mode' = 'earliest-offset'
);
-- JDBC connector
CREATE TABLE jdbc_accounts (
account_id STRING, holder_name STRING, balance DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:postgresql://db.bank.internal:5432/nbi',
'table-name' = 'accounts',
'username' = '${JDBC_USER}',
'password' = '${JDBC_PASS}'
);
-- Filesystem connector
CREATE TABLE file_reports (
branch_id STRING, total DECIMAL(15, 2), report_date DATE
) WITH (
'connector' = 'filesystem',
'path' = 's3://nbi-reports/daily/',
'format' = 'parquet'
);
What are the format options?
The format property tells the connector how to encode and decode each record's bytes.
-- JSON format
CREATE TABLE json_events (
event_id STRING, payload STRING
) WITH ('connector' = 'kafka', 'topic' = 'events', 'format' = 'json',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092');
-- Avro format with schema registry
CREATE TABLE avro_transactions (
txn_id STRING, amount DECIMAL(15, 2)
) WITH ('connector' = 'kafka', 'topic' = 'avro-txns',
'format' = 'avro-confluent',
'avro-confluent.url' = 'http://schema-registry.bank.internal:8081',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092');
-- CSV format
CREATE TABLE csv_import (
account_id STRING, balance DECIMAL(15, 2)
) WITH ('connector' = 'filesystem', 'path' = '/data/import/', 'format' = 'csv');
How does the watermark work?
The WATERMARK clause enables event-time processing. It declares which column is the event timestamp and how far behind the watermark trails. Without it, no downstream event-time window or temporal join can fire.
-- Watermark trails 5 seconds behind max observed event_time
CREATE TABLE loan_payments (
payment_id STRING,
loan_id STRING,
amount DECIMAL(15, 2),
paid_at TIMESTAMP(3),
WATERMARK FOR paid_at AS paid_at - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'nbi-loan-payments',
'format' = 'json',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092');
-- Now event-time windows work:
SELECT loan_id,
TUMBLE_START(paid_at, INTERVAL '1' HOUR) AS window_start,
SUM(amount) AS hourly_payments
FROM loan_payments
GROUP BY loan_id, TUMBLE(paid_at, INTERVAL '1' HOUR);
What are metadata columns?
Metadata columns expose connector-specific details like the Kafka partition, offset, or timestamp.
CREATE TABLE transactions_with_meta (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
kafka_partition INT METADATA FROM 'partition' VIRTUAL,
kafka_offset BIGINT METADATA FROM 'offset' VIRTUAL,
kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-transactions',
'format' = 'json',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092'
);
-- Query metadata for debugging
SELECT txn_id, kafka_partition, kafka_offset
FROM transactions_with_meta
WHERE account_id = 'ACCT-100';
What is a primary key and when is it needed?
A primary key declared as NOT ENFORCED is required for upsert sinks. It tells the sink which row to replace on updates.
-- Upsert sink needs a primary key
CREATE TABLE account_totals (
account_id STRING,
total_amount DECIMAL(15, 2),
txn_count BIGINT,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'nbi-account-totals',
'key.format' = 'json',
'value.format' = 'json'
);
INSERT INTO account_totals
SELECT account_id, SUM(amount), COUNT(*)
FROM transactions
GROUP BY account_id;
How do you use a table as both source and sink?
A table's usability as source or sink depends on the connector. Kafka and JDBC can be both. Filesystem supports both with partitioning.
-- Read from Kafka source, transform, write to JDBC sink
INSERT INTO jdbc_account_summary
SELECT account_id, SUM(amount) AS total, COUNT(*) AS txn_count
FROM kafka_transactions
GROUP BY account_id;
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class SourceAndSinkDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// Register source and sink tables via DDL
tEnv.executeSql("CREATE TABLE source_txns (...) WITH ('connector'='kafka', ...)");
tEnv.executeSql("CREATE TABLE sink_summary (...) WITH ('connector'='jdbc', ...)");
// Pipeline: read from Kafka, aggregate, write to JDBC
tEnv.executeSql(
"INSERT INTO sink_summary " +
"SELECT account_id, SUM(amount), COUNT(*) " +
"FROM source_txns GROUP BY account_id"
);
}
}
What should a senior engineer know?
- The
scan.startup.modefor Kafka controls whether you read from earliest, latest, or a timestamp, which matters for reprocessing. - Externalize connection secrets rather than hardcoding them in DDL. Use environment variables or a secret manager.
- Watermark definition belongs on the source table because that is where event time enters the pipeline.
- Some connectors use
key.formatandvalue.formatseparately for keyed Kafka topics. - Register tables in a persistent catalog so definitions are reusable across jobs and survive restarts.
-- Senior tip: use scan.startup.mode for reprocessing
CREATE TABLE replay_transactions (
txn_id STRING, amount DECIMAL(15, 2), event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-transactions',
'format' = 'json',
'scan.startup.mode' = 'timestamp',
'scan.startup.timestamp-millis' = '1704067200000',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092'
);
-- Replays from Jan 1, 2024 00:00:00 UTC
Visual Overview
CREATE TABLE Structure CREATE TABLE table_name ( ┌──────────────────────────────────────────┐ │ Column Definitions │ │ txn_id STRING, │ │ amount DECIMAL(15, 2), │ │ event_time TIMESTAMP(3), │ │ │ │ Metadata Columns (optional) │ │ kafka_offset BIGINT METADATA VIRTUAL, │ │ │ │ Primary Key (optional, for upsert sinks)│ │ PRIMARY KEY (account_id) NOT ENFORCED, │ │ │ │ Watermark (for event-time processing) │ │ WATERMARK FOR event_time AS │ │ event_time - INTERVAL '5' SECOND │ └──────────────────────────────────────────┘ ) WITH ( ┌──────────────────────────────────────────┐ │ Connector: kafka / jdbc / filesystem │ │ Format: json / avro / csv │ │ Connector-specific options │ └──────────────────────────────────────────┘ );
Watermark and Event-Time Flow
Source (with WATERMARK):
┌──────────────────────┐
│ event_time = 10:00 │──▶ watermark = 09:55 (5s behind)
│ event_time = 10:03 │──▶ watermark = 09:58
│ event_time = 10:07 │──▶ watermark = 10:02
└──────────────────────┘
│
▼
┌──────────────────────┐
│ TUMBLE(1 HOUR) │
│ Window 09:00-10:00 │
│ closes when watermark│
│ passes 10:00 │
└──────────────────────┘
Key Takeaway
In Flink SQL you define a table with CREATE TABLE that has three essential parts: the schema, an optional watermark, and a WITH clause. The schema lists columns with SQL types and can include metadata columns and a primary key. The WITH clause binds the table to an external system through the connector property (kafka, jdbc, filesystem) and the format property (json, avro, csv). Together they completely describe how bytes on the wire become rows.
The watermark clause enables event-time processing. It declares which column is the event timestamp and how far behind the watermark trails. Without it on the source, no downstream window or temporal join can fire. A primary key declared as NOT ENFORCED is required for upsert sinks so each write replaces the right row.
At the National Bank of India we defined the transactions source with the kafka connector and json format, with a watermark on event_time trailing five seconds so event-time windows closed correctly. We defined the account_totals sink with the jdbc connector and a primary key on account_id so aggregation results upserted correctly. The whole pipeline was just an INSERT INTO selecting a GROUP BY from the source.
Remember one sentence: CREATE TABLE's WITH clause is the single seam between declarative SQL and the external world, binding a connector, format, and watermark so Flink can read from and write to any system.
---