4. Flink Table & SQL API 入门：动态表概念、TableEnvironment创建、Kafka流表定义、简单SQL查询

在贷后预警场景中，实时数据流（如还款流水、客户行为事件）需要被持续计算。Flink Table & SQL API 提供了声明式的编程接口，让开发者可以用 SQL 表达流处理逻辑，大幅降低开发门槛。本章将围绕动态表的核心概念、环境创建、Kafka 流表定义以及基础 SQL 查询展开，为后续预警规则实现打下基础。

4.1 动态表（Dynamic Table）核心概念

传统数据库中的表是静态的，数据插入后不会自动变化。Flink 中的动态表则随时间不断演化——新数据到达时，表的内容会持续追加或更新。动态表是连接流处理与 SQL 的桥梁：

• 流 → 动态表：将无界数据流转换为带 Schema 的动态表，每条记录对应表的一行。
• 动态表 → 流：对动态表执行连续查询（Continuous Query），结果也是一个动态表，可再转换为 DataStream 输出。
• 查询连续性：SQL 查询在动态表上持续运行，每次数据变更都会触发结果更新，实现实时计算。

4.2 TableEnvironment 创建

TableEnvironment 是 Flink Table & SQL API 的入口，负责注册表、执行 SQL 查询、管理 UDF 等。根据执行环境的不同，创建方式分为两种：

在贷后预警项目中，我们使用流处理模式，推荐通过 EnvironmentSettings 创建，以便统一配置 Planner 和运行时行为：

// 1. 创建 StreamExecutionEnvironment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 2. 设置 EnvironmentSettings（使用 Blink Planner，流模式）
EnvironmentSettings settings = EnvironmentSettings.newInstance()
    .inStreamingMode()
    .build();

// 3. 创建 StreamTableEnvironment
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

4.3 Kafka 流表定义

贷后预警的实时数据通常来自 Kafka。Flink 通过 Kafka Connector 将 Topic 映射为动态表，定义时需要指定 Schema、连接参数以及 Format（如 JSON、CSV）。

以下示例定义了一个名为 repayment_events 的 Kafka 流表，用于接收还款事件：

tableEnv.executeSql(
    "CREATE TABLE repayment_events (" +
    "  event_time STRING, " +
    "  customer_id STRING, " +
    "  loan_id STRING, " +
    "  amount DECIMAL(10, 2), " +
    "  status STRING, " +
    "  proc_time AS PROCTIME() " +   // 处理时间属性
    ") WITH (" +
    "  'connector' = 'kafka', " +
    "  'topic' = 'repayment-topic', " +
    "  'properties.bootstrap.servers' = 'localhost:9092', " +
    "  'properties.group.id' = 'flink-warning-group', " +
    "  'scan.startup.mode' = 'latest-offset', " +
    "  'format' = 'json', " +
    "  'json.fail-on-missing-field' = 'false', " +
    "  'json.ignore-parse-errors' = 'true' " +
    ")"
);

4.4 简单 SQL 查询：SELECT、WHERE、GROUP BY

定义好 Kafka 流表后，即可使用 SQL 进行实时查询。以下展示三种基础操作：

4.4.1 SELECT 投影与字段过滤

// 查询所有字段
Table result = tableEnv.sqlQuery("SELECT * FROM repayment_events");

// 查询特定字段，并重命名
Table selected = tableEnv.sqlQuery(
    "SELECT customer_id, amount AS repay_amount, status FROM repayment_events"
);

4.4.2 WHERE 条件过滤

// 筛选逾期状态（status = 'OVERDUE'）且金额大于 1000 的事件
Table filtered = tableEnv.sqlQuery(
    "SELECT customer_id, amount, event_time " +
    "FROM repayment_events " +
    "WHERE status = 'OVERDUE' AND amount > 1000"
);

4.4.3 GROUP BY 聚合统计

// 按客户分组，统计每个客户的累计还款金额和事件次数
Table aggregated = tableEnv.sqlQuery(
    "SELECT customer_id, " +
    "       SUM(amount) AS total_repay, " +
    "       COUNT(*) AS event_count " +
    "FROM repayment_events " +
    "GROUP BY customer_id"
);

4.5 完整示例：贷后预警简单查询

将上述知识点串联，实现一个完整的贷后预警 SQL 查询：从 Kafka 读取还款事件，过滤出逾期记录，按客户分组统计逾期总金额，并输出到控制台。

public class LoanWarningSQLDemo {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 2. 定义 Kafka 流表
        tableEnv.executeSql(
            "CREATE TABLE repayment_events (" +
            "  event_time STRING, " +
            "  customer_id STRING, " +
            "  loan_id STRING, " +
            "  amount DECIMAL(10, 2), " +
            "  status STRING, " +
            "  proc_time AS PROCTIME() " +
            ") WITH (" +
            "  'connector' = 'kafka', " +
            "  'topic' = 'repayment-topic', " +
            "  'properties.bootstrap.servers' = 'localhost:9092', " +
            "  'properties.group.id' = 'flink-warning-group', " +
            "  'scan.startup.mode' = 'latest-offset', " +
            "  'format' = 'json', " +
            "  'json.fail-on-missing-field' = 'false', " +
            "  'json.ignore-parse-errors' = 'true' " +
            ")"
        );

        // 3. 执行 SQL 查询：逾期预警统计
        Table warningResult = tableEnv.sqlQuery(
            "SELECT customer_id, " +
            "       SUM(amount) AS total_overdue_amount, " +
            "       COUNT(*) AS overdue_count " +
            "FROM repayment_events " +
            "WHERE status = 'OVERDUE' " +
            "GROUP BY customer_id"
        );

        // 4. 将结果表转换为 DataStream 并打印
        tableEnv.toChangelogStream(warningResult).print("预警结果");

        // 5. 执行作业
        env.execute("Flink 贷后预警 SQL 作业");
    }
}

4.6 本章小结

• 动态表是 Flink SQL 流处理的核心抽象，将无界流映射为持续变化的表。
• TableEnvironment 是 SQL 作业的入口，推荐使用 EnvironmentSettings 创建流模式环境。
• Kafka 流表通过 DDL 定义，需指定 Schema、连接参数和 Format，支持 JSON 等常见格式。
• 简单 SQL 查询（SELECT、WHERE、GROUP BY）可直接应用于动态表，实现实时过滤与聚合。
• 贷后预警场景中，WHERE status='OVERDUE' 配合 GROUP BY customer_id 是最基础的预警统计模式。