Flink Table & SQL

Apache Flink 提供了强大的 Table API 和 SQL 接口，用于统一处理批数据和流数据。它们为开发者提供了类 SQL 的编程方式，简化了复杂的数据处理逻辑，并支持与外部系统集成。

🧩 一、Flink Table & SQL 核心概念

💻 二、Flink Table API 和 SQL 的优势

📦 三、核心组件说明

1. TableEnvironment

• 是操作 Table 和 SQL 的入口
• 负责注册表、执行查询、管理元数据等

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

2. DataStream ↔ Table 转换

示例：DataStream 转 Table

DataStream<Tuple2<String, Integer>> dataStream = env.fromElements(
    Tuple2.of("a", 1), Tuple2.of("b", 2));

// 将 DataStream 转换为 Table
Table table = tEnv.fromDataStream(dataStream);

// 注册为临时表
tEnv.createTemporaryView("myTable", dataStream);

示例：Table 转 DataStream

Table resultTable = tEnv.sqlQuery("SELECT * FROM myTable WHERE f1 > 1");
DataStream<Row> resultStream = tEnv.toDataStream(resultTable);

3. Flink SQL 查询

示例：使用 SQL 查询统计结果

// 创建临时表
tEnv.executeSql(
    "CREATE TABLE MyKafkaSource (" +
    "  user STRING," +
    "  url STRING," +
    "  ts BIGINT" +
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'format' = 'json'" +
    ")"
);

// 执行 SQL 查询
Table result = tEnv.sqlQuery("SELECT user, COUNT(*) AS cnt FROM MyKafkaSource GROUP BY user");

// 转换为 DataStream 并输出
tEnv.toDataStream(result).print();

env.execute();

🧪 四、Java 示例：完整的 Table API + SQL 使用案例

✅ 功能：

从 Kafka 读取日志数据，按用户分组统计访问次数

📁 依赖建议（pom.xml）

<dependencies>
    <!-- Flink Core -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.17.1</version>
    </dependency>

    <!-- Flink Streaming -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.17.1</version>
    </dependency>

    <!-- Flink Table API & SQL -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge</artifactId>
        <version>1.17.1</version>
    </dependency>

    <!-- Kafka Connector -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>1.17.1</version>
    </dependency>

    <!-- JSON Format -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>1.17.1</version>
    </dependency>
</dependencies>

🧱 五、完整 Java 示例代码

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

public class FlinkTableAndSQLEntry {

    public static void main(String[] args) throws Exception {
        // 1. 初始化流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 创建 TableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 3. 创建 Kafka Source 表（模拟从 Kafka 读取日志）
        tEnv.executeSql(
            "CREATE TABLE KafkaLog (" +
            "  user STRING," +
            "  url STRING," +
            "  ts BIGINT" +
            ") WITH (" +
            "  'connector' = 'kafka'," +
            "  'topic' = 'user_log'," +
            "  'properties.bootstrap.servers' = 'localhost:9092'," +
            "  'properties.group.id' = 'flink-sql-group'," +
            "  'format' = 'json'" +
            ")"
        );

        // 4. 创建 Sink 表（控制台输出）
        tEnv.executeSql(
            "CREATE TABLE ConsoleSink (" +
            "  user STRING," +
            "  cnt BIGINT" +
            ") WITH (" +
            "  'connector' = 'print'" +
            ")"
        );

        // 5. 使用 SQL 编写业务逻辑
        tEnv.executeSql(
            "INSERT INTO ConsoleSink " +
            "SELECT user, COUNT(*) AS cnt " +
            "FROM KafkaLog " +
            "GROUP BY user"
        );
    }
}

📊 六、SQL 查询示例汇总

⏱️ 七、时间属性与窗口聚合

示例：定义事件时间并使用滚动窗口

-- 定义带有事件时间的表
CREATE TABLE EventTable (
    user STRING,
    url STRING,
    ts BIGINT,
    WATERMARK FOR ts AS ts - 1000 -- 定义水印
) WITH (...);

-- 使用滚动窗口进行统计
SELECT 
    TUMBLE_END(ts, INTERVAL '5' SECOND) AS window_end,
    user,
    COUNT(*) AS cnt
FROM EventTable
GROUP BY TUMBLE(ts, INTERVAL '5' SECOND), user;

📁 八、连接器（Connector）配置示例

1. Kafka Source

CREATE TABLE KafkaSource (
    user STRING,
    url STRING,
    ts BIGINT
) WITH (
    'connector' = 'kafka',
    'topic' = 'input-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'flink-sql-group',
    'format' = 'json'
);

2. MySQL Sink

CREATE TABLE MysqlSink (
    user STRING,
    cnt BIGINT
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/mydb',
    'table-name' = 'user_access_log'
);

📈 九、Flink SQL + Table API 的典型应用场景

🧠 十、Table API vs SQL

✅ 十一、总结

🧩 十二、扩展学习方向

如果你希望我为你演示以下内容，请继续提问：

• 自定义函数（UDF、UDAF、UDTF）
• Kafka + MySQL 实时同步方案
• 基于 Hive 的批处理 SQL 作业
• 使用 PyFlink 实现 SQL 作业
• 使用 WITH 子句定义临时表
• 使用 LATERAL TABLE 调用 UDTF
• 使用 MATCH_RECOGNIZE 实现 CEP 模式匹配

📌 一句话总结：

Flink Table API 和 SQL 提供了一种统一的批流一体编程模型，适合数据仓库、实时分析、ETL、风控等多种大数据处理场景。