Table API and SQL Overview

Flink provides multiple API layers for stream processing. The core DataStream API serves as the foundation for building streaming applications, while lower-level Process Functions offer access to event time information, timers, and custom state management for stateful stream processing. Although these APIs are highly flexible and can theoretically handle all streaming scenarios, many organizations prefer SQL-based approaches for improved development efficiency.

Similar to other big data frameworks likee Hive and Spark, Flink offers SQL support through its Table API layer. This layer elevates the processing abstraction from DataStreams to Tables, transforming data processing operations into table transformations.

The Table API and SQL functionality underwent significant improvements after Flink 1.9 with the integration of Alibaba's Blink engine. Continuous development led to feature maturity by Flink 1.12, though even the latest versions continue to evolve with interface adjustments.

Getting Started

Using Table API and Flink SQL involves obtaining a Table instance and applying either Table API operations or SQL queries.

Required Dependencies

To use Table API in your code, include the following bridge dependency:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

This Java bridge connects Table API with the underlying DataStream API. For local IDE execution, additional dependencies are required:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

The blink planner provides the runtime environment and generates execution plans. For custom serialization formats, include:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
</dependency>

Basic Example

Consider processing event data containing user, URL, and timestamp fields, extracting only URL and user information. The SQL equivalent would be:

SELECT url, user FROM EventTable;

Java implementation:

public class TableExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        DataStream<Event> eventStream = env.fromElements(
            new Event("Alice", "./home", 1000L),
            new Event("Bob", "./cart", 1000L),
            new Event("Alice", "./prod?id=1", 5000L),
            new Event("Cary", "./home", 60000L),
            new Event("Bob", "./prod?id=3", 90000L),
            new Event("Alice", "./prod?id=7", 105000L)
        );
        
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table eventTable = tableEnv.fromDataStream(eventStream);
        Table resultTable = tableEnv.sqlQuery("SELECT url, user FROM " + eventTable);
        tableEnv.toDataStream(resultTable).print();
        env.execute();
    }
}

The programming model follows these steps:

1. Create stream execution environment
2. Obtain input data stream
3. Create table environment
4. Convert data stream to table
5. Apply query operations
6. Convert result table back to stream for output

Table API alternative:

Table resultTable = eventTable.select($("url"), $("user"));

Core API Components

Program Structure

Table API programs follow a pattern similar to DataStream API:

// 1. Create table environment
TableEnvironment tableEnv = ...;

// 2. Create input table
tableEnv.executeSql("CREATE TEMPORARY TABLE inputTable ... WITH ('connector' = ...)");

// 3. Create output table
tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ('connector' = ...)");

// 4. Execute query transformations
Table table1 = tableEnv.sqlQuery("SELECT ... FROM inputTable...");
Table table2 = tableEnv.from("inputTable").select(...);

// 5. Write results to output table
TableResult result = table1.executeInsert("outputTable");

Table Environment

TableEnvironment manages:

• Catalog and table registration
• SQL query execution
• User-defined function registration
• DataStream and Table conversions

Creating a table environment:

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();
TableEnvironment tableEnv = TableEnvironment.create(settings);

Or using stream environment:

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

Table Creation

Tables can be created through connectors or as virtual tables:

// Connector table
tableEnv.executeSql("CREATE TEMPORARY TABLE MyTable ... WITH ('connector' = ...)");

// Virtual table from query
Table newTable = tableEnv.sqlQuery("SELECT ... FROM MyTable ...");
tableEnv.createTemporaryView("NewTable", newTable);

Query Execution

Two query approaches are available:

// SQL approach
Table result = tableEnv.sqlQuery(
    "SELECT user, url FROM EventTable WHERE user = 'Alice'");

// Table API approach
Table result = eventTable
    .where($("user").isEqual("Alice"))
    .select($("url"), $("user"));

Output Operations

tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ('connector' = ...)");
Table result = ...;
result.executeInsert("OutputTable");

Stream-Table Conversions

Converting Table to DataStream:

// For append-only tables
tableEnv.toDataStream(resultTable).print();

// For updating tables
tableEnv.toChangelogStream(resultTable).print();

Converting DataStream to Table:

Table eventTable = tableEnv.fromDataStream(eventStream);
// With field selection and renaming
Table eventTable2 = tableEnv.fromDataStream(
    eventStream, $("timestamp").as("ts"), $("url"));

Dynamic Tables and Continuous Queries

In stream processing, tables evolve continuously as new data arrives, forming dynamic tables. Queries on these tables become continuous queries that produce updated results over time.

The stream processing pipeline involves:

1. Converting streams to dynamic tables
2. Applying continuous queries
3. Converting result tables back to streams

Query types include:

• Update queries: Results change with new data (require changelog streams)
• Append queries: Only insert operations (can use regular data streams)

Time Attributes and Windows

Time-based operations require defining time semantics through time attributes, which become part of the table schema.

Event Time

Defined in DDL:

CREATE TABLE EventTable(
    user STRING,
    url STRING,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (...);

Or during stream conversion:

Table eventTable = tableEnv.fromDataStream(
    eventStream, $("user"), $("url"), $("ts").rowtime());

Processing Time

CREATE TABLE EventTable(
    user STRING,
    url STRING,
    pt AS PROCTIME()
) WITH (...);

Or during stream conversion:

Table eventTable = tableEnv.fromDataStream(
    eventStream, $("user"), $("url"), $("pt").proctime());

Window Operations

Window table-valued functions:

// Tumbling window
TUMBLE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR)

// Sliding window
HOP(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '1' HOUR)

// Cumulative window
CUMULATE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR, INTERVAL '1' DAY)

Aggregation Queries

Group Aggregation

Table result = tableEnv.sqlQuery(
    "SELECT user, COUNT(url) AS cnt FROM EventTable GROUP BY user");

Window Aggregation

Table result = tableEnv.sqlQuery(
    "SELECT user, window_end, COUNT(url) AS cnt " +
    "FROM TABLE(TUMBLE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR)) " +
    "GROUP BY user, window_start, window_end");

Over Aggregation

Table result = tableEnv.sqlQuery(
    "SELECT user, ts, COUNT(url) OVER (" +
    "PARTITION BY user ORDER BY ts " +
    "RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW) AS cnt " +
    "FROM EventTable");

Join Operations

Flink SQL supports various join types including regular joins, interval joins, temporal joins, and lookup joins for different streaming scenarios.

Regular Join

SELECT * 
FROM table1 
INNER JOIN table2 
ON table1.id = table2.id;

Interval Join

SELECT *
FROM table1 
LEFT JOIN table2 
ON table1.id = table2.id 
AND table1.ts BETWEEN table2.ts - INTERVAL '1' HOUR AND table2.ts;

Temporal Join

SELECT *
FROM orders 
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
ON orders.currency = currency_rates.currency;

Lookup Join

SELECT *
FROM show_log s
LEFT JOIN user_profile FOR SYSTEM_TIME AS OF s.proctime AS u
ON s.user_id = u.user_id;