Apache Flink Table API — Unified Batch and Stream Processing with SQL

Why the Table API Sits Above DataStream API

Apache Flink offers two primary programming models. The DataStream API gives you full control over operators, state backends, and watermark strategies — it is the right choice when you need custom business logic or connectors that only exist at the low level. The DataStream API is covered in depth in the Flink streaming analytics guide, including stateful processing, custom watermark generators, and exactly-once sink patterns — the Table API builds on top of it, replacing boilerplate operator chains with a relational model that the Flink SQL planner can optimize automatically.

The Table API and SQL share the same planner: a query written as SQL strings and one written with the Java/Python Table DSL compile to identical execution plans. The planner applies optimizations — predicate pushdown into connectors, join reordering, mini-batch buffering for stateful aggregations — that are difficult or impossible to express manually in DataStream code. For most data engineering use cases (Kafka ingest, windowed aggregations, stream enrichment, lakehouse writes), the Table API delivers equivalent or better throughput with dramatically less code.

Project Setup and Table Environment

The Table API is available for Java, Scala, and Python. The Java and Scala flavors are production-grade; PyFlink supports the full Table API with some limitations on UDF performance. The Maven dependencies below cover streaming mode with Kafka and Parquet — the most common production combination.

<!-- pom.xml — core Flink Table API dependencies for Flink 1.20 -->
<properties>
  <flink.version>1.20.0</flink.version>
  <kafka.connector.version>3.3.0-1.20</kafka.connector.version>
</properties>

<dependencies>
  <!-- Table API bridge for streaming (includes TableEnvironment) -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
  </dependency>

  <!-- Blink SQL planner — required at runtime, not in fat-jar if using provided scope -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-loader</artifactId>
    <version>${flink.version}</version>
  </dependency>

  <!-- Kafka source + sink connector -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${kafka.connector.version}</version>
  </dependency>

  <!-- Parquet format for filesystem connector -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-parquet</artifactId>
    <version>${flink.version}</version>
  </dependency>

  <!-- JDBC connector for lookup (dimension) tables -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc</artifactId>
    <version>3.2.0-1.19</version>
  </dependency>
</dependencies>

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class OrderPipeline {
    public static void main(String[] args) throws Exception {

        // --- Streaming mode: unbounded sources (Kafka, CDC, etc.) ---
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // Checkpoint every 60s with exactly-once semantics
        env.enableCheckpointing(60_000);
        env.getCheckpointConfig().setCheckpointTimeout(120_000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // Mini-batch: buffer up to 5000 records or 5s before flushing aggregation state.
        // Reduces RocksDB read/write frequency by ~10x for high-cardinality GROUP BY.
        tableEnv.getConfig().set("table.exec.mini-batch.enabled", "true");
        tableEnv.getConfig().set("table.exec.mini-batch.allow-latency", "5 s");
        tableEnv.getConfig().set("table.exec.mini-batch.size", "5000");

        // Local-global aggregation: pre-aggregate on each subtask before shuffling.
        // Reduces network traffic for SUM/COUNT/MAX on high-cardinality keys.
        tableEnv.getConfig().set("table.optimizer.agg-phase-strategy", "TWO_PHASE");

        // --- Batch mode: bounded sources (Parquet, JDBC full extract) ---
        TableEnvironment batchEnv = TableEnvironment.create(
            EnvironmentSettings.newInstance().inBatchMode().build()
        );
        batchEnv.getConfig().set("table.exec.resource.default-parallelism", "8");

        // ... register sources, run queries, write sinks
    }
}

Defining Sources and Sinks with SQL DDL

The cleanest way to register connectors is SQL DDL. The CREATE TABLE statement maps directly to a connector implementation — no Java factory code, no custom source class. The WITH clause passes connector-specific properties. Schema inference is optional — declaring the schema explicitly is always safer in production because it catches type mismatches at compile time rather than at runtime.

-- Kafka source table with event-time watermarks.
-- The WATERMARK declaration tells Flink to use 'event_time' as the time attribute
-- and tolerate up to 5 seconds of out-of-order arrival before closing a window.
tableEnv.executeSql("""
    CREATE TABLE order_events (
        order_id    STRING,
        user_id     STRING,
        product_id  STRING,
        quantity    INT,
        amount      DECIMAL(12, 2),
        event_time  TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector'                       = 'kafka',
        'topic'                           = 'order-events',
        'properties.bootstrap.servers'    = 'kafka-broker:9092',
        'properties.group.id'             = 'flink-order-pipeline',
        'scan.startup.mode'               = 'latest-offset',
        'format'                          = 'json',
        'json.fail-on-missing-field'      = 'false',
        'json.ignore-parse-errors'        = 'true',
        'json.timestamp-format.standard'  = 'ISO-8601'
    )
""");

-- For multi-partition Kafka topics, enable partition discovery so new partitions
-- added after the job starts are consumed automatically.
-- 'properties.partition.discovery.interval.ms' = '60000'

-- Filesystem (S3/HDFS) sink with Parquet format and hourly partitioning.
-- 'sink.partition-commit.trigger' = 'partition-time' holds the partition open
-- until the watermark advances past partition_time + delay, then commits it.
tableEnv.executeSql("""
    CREATE TABLE orders_parquet_sink (
        order_id    STRING,
        user_id     STRING,
        product_id  STRING,
        amount      DECIMAL(12, 2),
        event_hour  STRING   -- derived partition column
    ) PARTITIONED BY (event_hour)
    WITH (
        'connector'                               = 'filesystem',
        'path'                                    = 's3a://my-bucket/data/orders/',
        'format'                                  = 'parquet',
        'parquet.compression'                     = 'SNAPPY',
        'sink.rolling-policy.rollover-interval'   = '10 min',
        'sink.rolling-policy.check-interval'      = '1 min',
        'sink.partition-commit.trigger'           = 'partition-time',
        'sink.partition-commit.delay'             = '1 h',
        'sink.partition-commit.policy.kind'       = 'success-file',
        'sink.partition-commit.watermark-time-zone' = 'UTC'
    )
""");

-- INSERT SELECT: derive the partition column from event_time.
-- DATE_FORMAT truncates to the hour for consistent hourly partition names.
tableEnv.executeSql("""
    INSERT INTO orders_parquet_sink
    SELECT
        order_id,
        user_id,
        product_id,
        amount,
        DATE_FORMAT(event_time, 'yyyy-MM-dd-HH') AS event_hour
    FROM order_events
""");

Window Aggregations — TUMBLE, HOP, and CUMULATE

Flink 1.13 introduced Table-Valued Functions (TVFs) for windowing. The TVF syntax — TABLE(TUMBLE(TABLE source, DESCRIPTOR(event_time), INTERVAL)) — replaces the older GROUP BY TUMBLE(event_time, INTERVAL) syntax and supports incremental window aggregation, which computes partial results per checkpoint rather than waiting for the full window to close. Use TVF syntax for all new Flink 1.18+ jobs.

-- TUMBLE window: fixed-size non-overlapping windows.
-- Each event belongs to exactly one window. Window closes when the watermark
-- passes window_end, emitting one result row per (key, window) combination.
SELECT
    product_id,
    window_start,
    window_end,
    COUNT(*)           AS order_count,
    SUM(amount)        AS total_revenue,
    AVG(amount)        AS avg_order_value,
    MAX(amount)        AS max_order
FROM TABLE(
    TUMBLE(TABLE order_events, DESCRIPTOR(event_time), INTERVAL '5' MINUTES)
)
GROUP BY product_id, window_start, window_end;

-- HOP (sliding) window: overlapping windows of fixed size advancing by a step.
-- Each event appears in SIZE/STEP windows. Here: 10-min window, 1-min slide
-- → each event contributes to 10 windows. Useful for rolling averages and
-- rate-of-change metrics without explicit LAG() computations.
SELECT
    user_id,
    window_start,
    window_end,
    COUNT(DISTINCT product_id)  AS unique_products_bought,
    SUM(amount)                 AS rolling_10m_spend
FROM TABLE(
    HOP(TABLE order_events, DESCRIPTOR(event_time),
        INTERVAL '1' MINUTE,    -- slide step
        INTERVAL '10' MINUTES)  -- window size
)
GROUP BY user_id, window_start, window_end;

-- CUMULATE window: growing window that resets at a fixed period.
-- Emits a result at every STEP interval within the current period.
-- Example: every hour, emit a running total since midnight.
-- Useful for intraday revenue dashboards without storing all day's events.
SELECT
    product_id,
    window_start,
    window_end,
    SUM(amount)    AS cumulative_revenue,
    COUNT(*)       AS cumulative_orders
FROM TABLE(
    CUMULATE(
        TABLE order_events,
        DESCRIPTOR(event_time),
        INTERVAL '1' HOUR,    -- step: emit every hour
        INTERVAL '1' DAY      -- max size: reset at midnight
    )
)
GROUP BY product_id, window_start, window_end;

Stream Joins — Interval Joins and Temporal Lookup Joins

Joining two unbounded streams is fundamentally different from joining two tables. Kafka Streams handles stream-stream joins with a time-bounded join window and dedicated state stores per join key — Flink's Table API expresses the same pattern as a standard SQL JOIN with a BETWEEN time condition that the planner translates into the appropriate stateful operator. The interval join bounds the state retention: events older than the window boundary are automatically evicted, preventing unbounded state growth.

-- Interval join: correlate order_events with shipment_events within 2 hours.
-- Flink retains order events for 2 hours to match incoming shipment events.
-- Events outside the interval are discarded — not late-emitted or buffered.
CREATE TABLE shipment_events (
    shipment_id  STRING,
    order_id     STRING,
    carrier      STRING,
    tracking_no  STRING,
    event_time   TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
    'connector'                     = 'kafka',
    'topic'                         = 'shipment-events',
    'properties.bootstrap.servers'  = 'kafka-broker:9092',
    'properties.group.id'           = 'flink-shipment-join',
    'scan.startup.mode'             = 'latest-offset',
    'format'                        = 'json'
);

SELECT
    o.order_id,
    o.user_id,
    o.amount,
    s.shipment_id,
    s.carrier,
    s.tracking_no,
    o.event_time  AS order_time,
    s.event_time  AS ship_time,
    TIMESTAMPDIFF(MINUTE, o.event_time, s.event_time) AS fulfillment_minutes
FROM order_events  AS o
JOIN shipment_events AS s
  ON  o.order_id    = s.order_id
  AND s.event_time BETWEEN o.event_time
                       AND o.event_time + INTERVAL '2' HOURS;

-- Temporal table join (lookup join): enrich stream with dimension data
-- from a JDBC table, looking up the record valid AT the event's event_time.
-- The dimension table must have a primary key; Flink caches lookups using
-- 'lookup.cache.max-rows' and 'lookup.cache.ttl' to avoid per-event DB queries.
CREATE TABLE product_catalog (
    product_id    STRING,
    product_name  STRING,
    category      STRING,
    unit_price    DECIMAL(12, 2),
    updated_at    TIMESTAMP(3),
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'connector'               = 'jdbc',
    'url'                     = 'jdbc:postgresql://postgres:5432/catalog',
    'table-name'              = 'products',
    'username'                = 'flink_reader',
    'password'                = '${POSTGRES_PASSWORD}',
    'lookup.cache.max-rows'   = '10000',
    'lookup.cache.ttl'        = '10 min',
    'lookup.max-retries'      = '3'
);

-- FOR SYSTEM_TIME AS OF performs a point-in-time lookup:
-- for each order event, retrieve the product record as it existed at event_time.
-- This is the correct pattern for slowly-changing dimensions in streaming ETL.
SELECT
    o.order_id,
    o.user_id,
    o.product_id,
    o.quantity,
    o.amount,
    p.product_name,
    p.category,
    p.unit_price,
    o.amount / o.quantity AS actual_unit_price,
    o.event_time
FROM order_events AS o
LEFT JOIN product_catalog FOR SYSTEM_TIME AS OF o.event_time AS p
  ON o.product_id = p.product_id;

Table API — Type-Safe Programmatic Style

The Table API DSL provides the same relational operations as SQL but through Java/Python method calls with IDE autocompletion and compile-time type checking. It compiles to the same execution plan as SQL — choose based on your team's preference. The programmatic API is easier to unit-test (mock a TableEnvironment with a bounded in-memory source) and to compose dynamically (build filter expressions from user input without string concatenation).

import org.apache.flink.table.api.*;
import static org.apache.flink.table.api.Expressions.*;

Table orders = tableEnv.from("order_events");

// --- Project and filter ---
Table highValueOrders = orders
    .select($("order_id"), $("user_id"), $("product_id"), $("amount"), $("event_time"))
    .filter($("amount").isGreater(lit(500.00)));

// --- GROUP BY aggregation (non-windowed / continuous) ---
// Warning: this accumulates unbounded state. Add a TTL or use windowed aggregation.
tableEnv.getConfig().set("table.exec.state.ttl", "24 h"); // evict state after 24h idle
Table perProductTotals = orders
    .groupBy($("product_id"))
    .select(
        $("product_id"),
        $("order_id").count().as("total_orders"),
        $("amount").sum().as("total_revenue"),
        $("amount").max().as("max_order")
    );

// --- Tumbling window via Table API builder ---
Table windowedRevenue = orders
    .window(
        Tumble.over(lit(5).minutes())
              .on($("event_time"))
              .as("w")
    )
    .groupBy($("product_id"), $("w"))
    .select(
        $("product_id"),
        $("w").start().as("window_start"),
        $("w").end().as("window_end"),
        $("amount").sum().as("revenue"),
        $("order_id").count().as("orders")
    );

// --- Join (static enrichment pattern) ---
Table catalog = tableEnv.from("product_catalog");
Table enriched = orders
    .join(catalog, $("product_id").isEqual(catalog.$("product_id")))
    .select(
        orders.$("order_id"),
        orders.$("user_id"),
        orders.$("amount"),
        catalog.$("product_name"),
        catalog.$("category")
    );

// --- Write to sink ---
windowedRevenue.executeInsert("orders_parquet_sink");

DataStream ↔ Table API Conversion

The StreamTableEnvironment bridges the two APIs. Converting a DataStream to a Table is the right move when your source connector is only available at the DataStream level (custom Kafka deserialization, Flink CDC, etc.) and you want to query its output with SQL. Converting back from Table to DataStream is needed when the sink connector is only available at the DataStream level or when you need custom downstream processing (e.g., a stateful process function after SQL windowing).

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

// --- DataStream → Table ---
// Custom Kafka source returning domain objects (not via SQL DDL)
DataStream<OrderEvent> orderStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-orders");

// Register as a temporary view, declaring the time attribute and watermark.
// The Schema.newBuilder() lets you override column names/types and add watermarks
// that cannot be expressed in the DataStream's POJO itself.
tableEnv.createTemporaryView(
    "order_events_raw",
    orderStream,
    Schema.newBuilder()
        .column("orderId",    DataTypes.STRING())
        .column("userId",     DataTypes.STRING())
        .column("productId",  DataTypes.STRING())
        .column("amount",     DataTypes.DECIMAL(12, 2))
        .columnByExpression("proc_time", "PROCTIME()")  // processing-time attribute
        .columnByMetadata("event_time", DataTypes.TIMESTAMP_LTZ(3), "timestamp", true)
        .watermark("event_time", "event_time - INTERVAL '5' SECOND")
        .build()
);

// --- Table → DataStream ---
// toDataStream: append-only stream — correct for INSERT-only result tables.
Table result = tableEnv.sqlQuery("""
    SELECT product_id, window_start, window_end, SUM(amount) AS revenue
    FROM TABLE(TUMBLE(TABLE order_events_raw, DESCRIPTOR(event_time), INTERVAL '5' MINUTES))
    GROUP BY product_id, window_start, window_end
""");

DataStream<Row> appendStream = tableEnv.toDataStream(result);
appendStream.print();

// toChangelogStream: retract/upsert stream — required for UPDATE or DELETE results
// (non-windowed GROUP BY, regular joins). Rows carry a RowKind flag (+I, -U, +U, -D).
DataStream<Row> changelogStream = tableEnv.toChangelogStream(perProductTotals);

// env.execute() is required when mixing Table API with DataStream API.
// Pure Table API jobs (only executeInsert / TableResult) do NOT need env.execute().
env.execute("Order Analytics Hybrid Pipeline");

Catalog Integration — Iceberg REST and Hive Metastore

Without a catalog, tables registered via CREATE TABLE are temporary — they exist only for the lifetime of the running job and are not visible to other jobs or Flink sessions. A catalog persists table definitions so they survive job restarts and can be queried from multiple jobs, Spark, Trino, or Hive. Apache Iceberg's REST Catalog is the most portable choice: it separates the catalog server from the storage layer and is supported natively by Flink, Spark, Trino, and DuckDB with no vendor lock-in.

// Register an Iceberg REST Catalog — tables persist across job restarts
// and are queryable from Spark, Trino, and DuckDB using the same endpoint.
tableEnv.executeSql("""
    CREATE CATALOG iceberg_prod WITH (
        'type'                = 'iceberg',
        'catalog-type'        = 'rest',
        'uri'                 = 'http://iceberg-rest-catalog:8181',
        'warehouse'           = 's3a://my-lakehouse/warehouse',
        'io-impl'             = 'org.apache.iceberg.aws.s3.S3FileIO',
        's3.endpoint'         = 'http://minio:9000',
        's3.path-style-access' = 'true',
        'client.region'       = 'us-east-1'
    )
""");

tableEnv.useCatalog("iceberg_prod");
tableEnv.useDatabase("analytics");

// Create a persistent Iceberg table — survives job restart
tableEnv.executeSql("""
    CREATE TABLE IF NOT EXISTS daily_order_summary (
        event_date     DATE,
        product_id     STRING,
        total_orders   BIGINT,
        total_revenue  DECIMAL(18, 2),
        avg_order      DECIMAL(12, 2),
        PRIMARY KEY (event_date, product_id) NOT ENFORCED
    ) WITH (
        'format-version'  = '2',
        'write.format.default' = 'parquet',
        'write.parquet.compression-codec' = 'snappy'
    )
""");

// Populate from streaming aggregation
tableEnv.executeSql("""
    INSERT INTO daily_order_summary
    SELECT
        CAST(window_start AS DATE)  AS event_date,
        product_id,
        COUNT(*)                    AS total_orders,
        SUM(amount)                 AS total_revenue,
        AVG(amount)                 AS avg_order
    FROM TABLE(
        TUMBLE(TABLE order_events, DESCRIPTOR(event_time), INTERVAL '1' DAY)
    )
    GROUP BY product_id, window_start, window_end
""");

// Hive Catalog — uses the Hive Metastore for schema persistence.
// Tables created here are visible in Hive, SparkSQL, and Presto/Trino.
// Requires flink-connector-hive and hive-exec on the classpath.
tableEnv.executeSql("""
    CREATE CATALOG hive_catalog WITH (
        'type'              = 'hive',
        'default-database'  = 'data_warehouse',
        'hive-conf-dir'     = '/opt/flink/conf/hive',
        'hadoop-conf-dir'   = '/etc/hadoop/conf'
    )
""");

// List all databases in the Hive catalog
tableEnv.useCatalog("hive_catalog");
tableEnv.executeSql("SHOW DATABASES").print();
tableEnv.executeSql("SHOW TABLES IN data_warehouse").print();

Production Configuration — Checkpointing, State Backend, and Tuning

Flink's fault-tolerance model depends on periodic checkpoints. For streaming Table API jobs with RocksDB-backed state (window aggregations, join buffers, GROUP BY accumulators), incremental checkpointing dramatically reduces checkpoint size — only the changed SST files are uploaded to S3, not the full state snapshot. A job with 100 GB of RocksDB state writes ~500 MB of incremental checkpoint data per minute rather than 100 GB per checkpoint.

# flink-conf.yaml (or flink-conf.properties in Flink 1.19+)

# --- Checkpointing ---
execution.checkpointing.interval: 60s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.min-pause: 30s       # minimum gap between checkpoints
execution.checkpointing.timeout: 120s
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

# --- State Backend (RocksDB for large state) ---
state.backend.type: rocksdb
state.backend.incremental: true              # incremental checkpoints — critical for large state
state.checkpoints.dir: s3a://my-bucket/flink-checkpoints/${JOB_NAME}
state.savepoints.dir: s3a://my-bucket/flink-savepoints/${JOB_NAME}

# RocksDB memory: managed mode lets Flink control per-slot memory
state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.fixed-per-slot: 512mb
state.backend.rocksdb.memory.write-buffer-ratio: 0.5
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

# --- Table planner optimizations ---
table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5 s
table.exec.mini-batch.size: 5000
table.optimizer.agg-phase-strategy: TWO_PHASE   # local-global aggregation

# State TTL: automatically evict idle keys from non-windowed GROUP BY
table.exec.state.ttl: 24 h

# --- JobManager High Availability (Kubernetes) ---
high-availability.type: kubernetes
high-availability.storageDir: s3a://my-bucket/flink-ha/${JOB_NAME}
kubernetes.cluster-id: order-analytics-cluster

# --- Metrics ---
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249

Production Checklist