4. Flink SQL 核心基础：环境搭建、Table & SQL API 入门、动态表与连续查询

好，咱们进入 Flink SQL 的世界。这一章是地基，我建议你认真过一遍。很多同学上来就写 SQL，结果环境配不对、表概念搞混，后面排查问题特别痛苦。我自己带团队时，踩过不少这种坑。

4.1 开发环境搭建：别让环境卡住你

先说环境。Flink 支持本地开发、集群部署，咱们课程里主要用本地模式跑例子。你只需要三样东西：JDK 8/11、Maven 3.6+、IDE（我习惯用 IntelliJ IDEA）。

创建一个 Maven 项目，引入核心依赖：

<properties>
    <flink.version>1.17.0</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-loader</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</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;

public class FlinkSQLDemo {
    public static void main(String[] args) {
        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 创建表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        System.out.println("Flink SQL 环境搭建成功！");
    }
}

跑一下，控制台没报错，说明环境通了。嗯，这里要注意：inStreamingMode() 表示我们工作在流模式下，这是实时宽表的核心场景。

4.2 Table & SQL API 入门：两种姿势，一个内核

Flink 提供了两套 API：Table API 和 SQL API。说白了，Table API 是 Java/Scala 的链式调用，SQL API 就是写标准 SQL。底层引擎是一样的。

我刚开始用的时候，总觉得 Table API 有点多余。后来发现，在复杂 ETL 场景下，Table API 的算子组合比 SQL 字符串拼接更灵活。举个例子：

4.2.1 用 Table API 注册表

// 从 DataStream 创建 Table
DataStream<Order> orderStream = env.addSource(...);
Table orderTable = tableEnv.fromDataStream(orderStream,
        $("orderId"),
        $("userId"),
        $("amount"),
        $("orderTime").rowtime());

// 注册为临时视图
tableEnv.createTemporaryView("orders", orderTable);

4.2.2 用 SQL 查询

Table result = tableEnv.sqlQuery(
    "SELECT userId, SUM(amount) AS totalAmount " +
    "FROM orders " +
    "GROUP BY userId"
);

你看，两种方式可以混用。我个人习惯：简单查询用 SQL，复杂多步骤用 Table API。为什么？SQL 可读性好，但调试起来麻烦；Table API 可以在 IDE 里一步步断点。

4.3 动态表与连续查询：流式 SQL 的灵魂

好，重点来了。传统数据库里，表是静态的——数据插入后就不变了。但 Flink SQL 里，表是动态的。什么意思？

你想想看，一个 Kafka 消息流源源不断进来，每来一条数据，表的内容就变了。Flink 把这种随时间变化的表叫做动态表（Dynamic Table）。而我们对动态表执行的 SQL，就是连续查询（Continuous Query）——它永远不会终止，每来一条数据，结果就更新一次。

我记得第一次理解这个概念时，脑子有点绕。画个图就清楚了：

4.4 动态表的三种更新模式

动态表不是随便变的，它有明确的更新模式。我整理了一张表，你对照着看：

4.5 时间属性：流式 SQL 的命脉

动态表离不开时间。Flink SQL 支持三种时间语义：

• 事件时间（Event Time）：数据本身携带的时间戳。最准确，但需要处理乱序。
• 处理时间（Processing Time）：Flink 算子所在机器的当前时间。简单，但不准确。
• 摄入时间（Ingestion Time）：数据进入 Flink 的时间。介于两者之间。

我建议你优先使用事件时间。为什么？实时宽表要保证数据一致性，处理时间在重算或回溯时结果会变。举个例子：

// 定义事件时间
Table orders = tableEnv.fromDataStream(orderStream,
    $("orderId"),
    $("userId"),
    $("amount"),
    $("orderTime").rowtime());  // 指定事件时间

// 基于事件时间的滚动窗口
Table result = tableEnv.sqlQuery(
    "SELECT " +
    "  userId, " +
    "  TUMBLE_END(orderTime, INTERVAL '1' MINUTE) AS windowEnd, " +
    "  SUM(amount) AS totalAmount " +
    "FROM orders " +
    "GROUP BY userId, TUMBLE(orderTime, INTERVAL '1' MINUTE)"
);

4.6 本章小结

这一章我们搭好了 Flink SQL 的开发环境，理解了 Table API 和 SQL API 的关系，最重要的是搞懂了动态表和连续查询这两个核心概念。说白了，实时宽表就是在一张不断变化的动态表上，跑一个永不停止的 SQL 查询。