📚 课程目录第4章 Flink Table & SQL API

4. Flink Table & SQL API：TableEnvironment、动态表、流式SQL查询、与DataStream互转

在保险精算坏样本归因场景中，数据源通常是实时流入的理赔事件流（DataStream）。Flink Table & SQL API 提供了声明式的编程范式，允许精算师使用标准 SQL 对动态变化的流数据进行查询、聚合与归因分析。本章将深入讲解 TableEnvironment 的核心作用、动态表的概念、流式 SQL 的执行机制，以及 Table 与 DataStream 之间的双向转换方法。

4.1 TableEnvironment：流式SQL的执行上下文

TableEnvironment 是 Flink Table & SQL API 的入口，负责管理表元数据、注册自定义函数、执行 SQL 查询以及配置执行环境。在流处理模式下，它通过 StreamTableEnvironment 实现。

组件	说明	精算场景示例
EnvironmentSettings	定义执行模式（流/批）、时间特性、状态后端等	设置 `inStreamingMode()` 处理实时理赔流
StreamTableEnvironment	基于 StreamExecutionEnvironment 创建，提供 SQL 执行能力	`StreamTableEnvironment.create(env)`
TableCatalog	管理表注册、视图、函数	注册 Kafka 主题为动态表
SQL 查询	执行 SELECT、JOIN、GROUP BY 等标准 SQL	按产品维度聚合坏样本率

  精算实践提示： 在坏样本归因作业中，建议将 TableEnvironment 的 sql.exec.source.idle-timeout 参数设置为 5 秒，避免因上游 Kafka 分区短暂无数据导致 watermark 停滞，从而影响滑动窗口聚合的准确性。

// 创建流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建 Table 环境，启用流模式
EnvironmentSettings settings = EnvironmentSettings.newInstance()
    .inStreamingMode()
    .build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

// 注册 Kafka 表（理赔事件流）
tableEnv.executeSql(
    "CREATE TABLE claim_events (" +
    "  claim_id STRING," +
    "  product_code STRING," +
    "  claim_amount DOUBLE," +
    "  is_fraud BOOLEAN," +
    "  event_time TIMESTAMP(3)," +
    "  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND" +
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'topic' = 'claim-input'," +
    "  'properties.bootstrap.servers' = 'localhost:9092'," +
    "  'format' = 'json'" +
    ")"
);

4.2 动态表：流数据的逻辑视图

动态表（Dynamic Table）是 Flink 对流数据在逻辑层面的抽象。与静态批处理表不同，动态表会随时间持续变化——新记录到达时，表的内容被追加或更新。在保险精算中，每一条新理赔事件都会实时反映在动态表中。

关键理解： 动态表不是物理存储，而是对流数据的逻辑抽象。每条新理赔记录到达时，动态表的内容立即变化，SQL 查询的结果也随之更新。这种“持续查询”机制是 Flink 实现实时归因分析的核心。

4.3 流式SQL查询：持续计算与结果刷新

在动态表上执行的 SQL 查询被称为“持续查询”（Continuous Query）。与传统批处理 SQL 只执行一次不同，流式 SQL 查询会持续监听动态表的变化，并在每次数据到达时增量更新计算结果。

4.3.1 查询类型与输出模式

查询类型	SQL 示例	输出模式	精算应用
追加查询	`SELECT * FROM claim_events WHERE amount > 10000`	Append（只追加）	筛选大额理赔事件
更新查询	`SELECT product_code, COUNT(*) FROM claim_events GROUP BY product_code`	Retract / Upsert	按产品统计坏样本数
窗口聚合	`SELECT TUMBLE_END(event_time, INTERVAL '1' HOUR), product_code, AVG(claim_amount) FROM claim_events GROUP BY TUMBLE(event_time, INTERVAL '1' HOUR), product_code`	Append（窗口结束输出）	每小时计算平均理赔额

精算归因注意： 使用 GROUP BY 进行无窗口的无限聚合时，Flink 会维护一个不断增长的状态。对于坏样本归因中的产品维度聚合，建议配合 TTL（状态生存时间） 配置，避免状态无限膨胀。例如：tableEnv.getConfig().setIdleStateRetention(Duration.ofHours(24))。

// 流式 SQL：按产品代码统计坏样本数量（持续查询）
Table resultTable = tableEnv.sqlQuery(
    "SELECT " +
    "  product_code, " +
    "  COUNT(*) AS total_claims, " +
    "  SUM(CASE WHEN is_fraud = true THEN 1 ELSE 0 END) AS fraud_count, " +
    "  ROUND(SUM(CASE WHEN is_fraud = true THEN 1 ELSE 0 END) * 1.0 / COUNT(*), 4) AS fraud_rate " +
    "FROM claim_events " +
    "GROUP BY product_code"
);

// 将结果表转换为 DataStream 输出到外部系统
tableEnv.toRetractStream(resultTable, Row.class)
    .filter(row -> row.f1 != null)  // 过滤 retract 消息
    .map(row -> row.f1.toString())
    .print();

4.4 Table 与 DataStream 互转

在保险精算坏样本归因的工程实践中，经常需要在 Table API 的声明式 SQL 和 DataStream API 的过程式处理之间切换。Flink 提供了双向转换方法，使开发者可以灵活组合两种编程范式。

4.4.1 DataStream → Table

将 DataStream 注册为 Table 或转换为 Table 对象，是使用 SQL 分析流数据的前提。

方法	说明	适用场景
`tableEnv.fromDataStream(stream)`	将 DataStream 直接转为 Table，字段按 POJO 属性或 Tuple 位置映射	简单转换，字段名自动推导
`tableEnv.fromDataStream(stream, Schema)`	指定字段名、时间戳、watermark 等 Schema 信息	需要显式定义事件时间和 watermark
`tableEnv.createTemporaryView(name, stream)`	注册为临时视图，后续 SQL 可直接引用	多表 JOIN 或复杂 SQL 查询

// 从 DataStream 创建 Table，并指定事件时间
DataStream<ClaimEvent> claimStream = env
    .addSource(new FlinkKafkaConsumer<>(...))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<ClaimEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, ts) -> event.getEventTime())
    );

// 方式一：直接转换，自动推导字段
Table claimTable = tableEnv.fromDataStream(claimStream);

// 方式二：指定 Schema，显式声明事件时间
Table claimTableWithSchema = tableEnv.fromDataStream(
    claimStream,
    Schema.newBuilder()
        .column("claimId", DataTypes.STRING())
        .column("productCode", DataTypes.STRING())
        .column("claimAmount", DataTypes.DOUBLE())
        .column("isFraud", DataTypes.BOOLEAN())
        .column("eventTime", DataTypes.TIMESTAMP(3))
        .watermark("eventTime", "eventTime - INTERVAL '5' SECOND")
        .build()
);

// 方式三：注册为临时视图，供 SQL 直接查询
tableEnv.createTemporaryView("claim_view", claimStream,
    $("claimId"), $("productCode"), $("claimAmount"), $("isFraud"), $("eventTime").rowtime());

4.4.2 Table → DataStream

将 Table 转换回 DataStream 后，可以继续使用 DataStream API 进行自定义处理、侧输出、状态编程或与外部系统交互。

方法	输出类型	适用查询类型	精算场景
`tableEnv.toDataStream(table)`	DataStream<Row>	仅追加查询（Append）	筛选后的理赔事件流
`tableEnv.toChangelogStream(table)`	DataStream<Row>（含变更类型）	更新查询（Retract/Upsert）	聚合结果的变化流
`tableEnv.toRetractStream(table, typeInfo)`	DataStream<Tuple2<Boolean, T>>	更新查询（Retract 模式）	坏样本率实时更新

// 将聚合结果 Table 转换为 Retract Stream
Table fraudStats = tableEnv.sqlQuery(
    "SELECT product_code, COUNT(*) AS total, " +
    "SUM(CASE WHEN is_fraud THEN 1 ELSE 0 END) AS frauds " +
    "FROM claim_events GROUP BY product_code"
);

// 方式一：toRetractStream - 每条记录包含 (true=新增/更新, false=撤回)
DataStream<Tuple2<Boolean, Row>> retractStream =
    tableEnv.toRetractStream(fraudStats, Row.class);

retractStream
    .filter(tuple -> tuple.f0)  // 只保留新增/更新消息
    .map(tuple -> {
        Row row = tuple.f1;
        String product = row.getField(0).toString();
        long total = (long) row.getField(1);
        long frauds = (long) row.getField(2);
        double rate = total > 0 ? (double) frauds / total : 0.0;
        return product + " | 总量:" + total + " | 坏样本:" + frauds + " | 坏样本率:" + String.format("%.2f%%", rate * 100);
    })
    .print();

// 方式二：toChangelogStream - 更丰富的变更类型
DataStream<Row> changelogStream = tableEnv.toChangelogStream(fraudStats);
changelogStream.print();

4.5 精算场景综合示例：坏样本归因流水线

以下示例展示如何将 Table API 与 DataStream API 组合，构建一个完整的坏样本归因流水线：

// 1. 创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 2. 从 Kafka 读取理赔流（DataStream）
DataStream<ClaimEvent> claimStream = env
    .addSource(new FlinkKafkaConsumer<>("claim-input", new ClaimEventDeserializer(), props))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<ClaimEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner((event, ts) -> event.getEventTime())
    );

// 3. 注册为 Table 视图
tableEnv.createTemporaryView("claims", claimStream,
    $("claimId"), $("productCode"), $("claimAmount"),
    $("isFraud"), $("eventTime").rowtime());

// 4. 使用 SQL 进行坏样本归因聚合
Table attributionResult = tableEnv.sqlQuery(
    "SELECT " +
    "  productCode, " +
    "  TUMBLE_END(eventTime, INTERVAL '10' MINUTE) AS windowEnd, " +
    "  COUNT(*) AS totalClaims, " +
    "  SUM(CASE WHEN isFraud THEN 1 ELSE 0 END) AS fraudClaims, " +
    "  ROUND(SUM(CASE WHEN isFraud THEN claimAmount ELSE 0 END) / " +
    "        SUM(CASE WHEN isFraud THEN 1 ELSE 0 END), 2) AS avgFraudAmount " +
    "FROM claims " +
    "GROUP BY productCode, TUMBLE(eventTime, INTERVAL '10' MINUTE)"
);

// 5. 转换为 DataStream 进行后处理（如关联外部维度表）
DataStream<Row> resultStream = tableEnv.toDataStream(attributionResult);

resultStream
    .map(row -> {
        String product = row.getField(0).toString();
        String windowEnd = row.getField(1).toString();
        long total = (long) row.getField(2);
        long frauds = (long) row.getField(3);
        double avgAmt = row.getField(4) != null ? (double) row.getField(4) : 0.0;
        double fraudRate = total > 0 ? (double) frauds / total : 0.0;
        return String.format("[%s] 产品:%s | 总量:%d | 坏样本:%d | 坏样本率:%.2f%% | 平均坏样本额:%.2f",
            windowEnd, product, total, frauds, fraudRate * 100, avgAmt);
    })
    .addSink(new FlinkKafkaProducer<>("attribution-output", new SimpleStringSchema(), props));

env.execute("坏样本归因分析作业");

架构要点： 上述流水线展示了“DataStream 接入 → Table SQL 分析 → DataStream 输出”的经典模式。精算师可以专注于 SQL 层的归因逻辑编写，而将数据接入、状态管理、外部系统交互等底层细节交给 DataStream API 处理。这种分层设计既保证了分析逻辑的声明式简洁性，又保留了过程式编程的灵活性。