欢迎光临
我们一直在努力
        找回密码
当前位置：171主机测评 > 服务器技术 > 正文

基于 Apache Flink 的实时计算平台：Flink SQL 与 Table API 双模式实战指南

2026-10-20 分类：服务器技术 阅读(3) 评论(0)
  • 示例工程
  • 大数据

【免费下载链接】flink-learning

flink learning blog. http://www.54tianzhisheng.cn/ 含 Flink 入门、概念、原理、实战、性能调优、源码解析等内容。涉及 Flink Connector、Metrics、Library、DataStream API、Table API & SQL 等内容的学习案例，还有 Flink 落地应用的大型项目案例（PVUV、日志存储、百亿数据实时去重、监控告警）分享。欢迎大家支持我的专栏《大数据实时计算引擎 Flink 实战与性能优化》

项目地址：
https://gitcode.com/gh_mirrors/fl/flink-learning

点击查看 免费下载

本文是 flink-learning 仓库中「实时计算平台建设」模块的技术详解，围绕 flink-learning-project-real-time-computing-platform 模块的两份核心作业展开：以 Flink SQL DDL 为骨架的 FlinkSqlPlatformJob，以及以 Table API 编程为骨架的 TableApiExampleJob。读者通过本文可以掌握如何用 Flink SQL 与 Table API 两种方式搭建实时计算任务（Kafka 接入、Watermark 定义、滚动窗口聚合、UV 去重、动态表与流互转），并理解两条技术路线在实时计算平台中的分工与取舍。

一、模块定位：实时计算平台的两种提交范式

在真实的实时计算平台中，用户提交任务的方式通常分为两类：一类是数据分析师直接书写 SQL，交由平台解析执行；另一类是平台开发者用编程方式（Java/Scala）实现复杂业务逻辑。本模块正是围绕这两种范式给出的参考实现，代码位于：

  • SQL 方式：FlinkSqlPlatformJob.java
  • Table API 方式：TableApiExampleJob.java

两个作业都基于 StreamTableEnvironment（流表桥接环境）构建，它是连接 DataStream API 与 Table/SQL API 的关键入口。模块依赖关系见 pom.xml：除了共享的 flink-learning-project-common 模块外，还引入了 flink-table-api-java-bridge（provided 作用域，由运行环境提供）。

需要说明的是，本仓库根目录 pom.xml 将 flink.version 统一声明为 1.20.3、Kafka 连接器版本为 3.4.0-1.20、Java 编译目标为 11，因此下面展示的代码均以 Flink 1.20 系列的新版 API（如 KafkaSource、KafkaSink）为准。

二、模式一：FlinkSqlPlatformJob——纯 SQL 构建实时计算任务

FlinkSqlPlatformJob 以「订单实时统计」为业务场景，完整演示了从建表、声明 Watermark、执行窗口聚合到写出结果的 SQL 全流程，其执行链路为：

创建 StreamTableEnvironment → DDL 创建 Kafka Source 表 → DDL 创建 Sink 表 → 执行 INSERT INTO 聚合 SQL

2.1 创建流表桥接环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.inStreamingMode()
.build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

要点说明：

  • env.setParallelism(4)：设定作业整体并行度为 4，实际生产应依据 Kafka 分区数动态调整；
  • EnvironmentSettings.newInstance().inStreamingMode()：显式声明流处理模式（非批模式），这是流式实时计算平台的标准配置；
  • StreamTableEnvironment.create(env, settings)：将 DataStream 执行环境包装为流表环境，此后 DDL 与 SQL 都通过 tableEnv.executeSql(…) 提交。

2.2 用 DDL 定义 Kafka Source 表（含 Watermark 策略）

平台中数据源通常由平台侧统一注册，用户只需引用表名即可。本示例直接通过 SQL DDL 在作业内注册 ods_order 订单源表：

CREATE TABLE ods_order (
order_id STRING,
user_id STRING,
product_name STRING,
category STRING,
price DOUBLE,
quantity INT,
total_amount DOUBLE,
province STRING,
create_time BIGINT,
— 从 BIGINT 时间戳生成事件时间属性
ts AS TO_TIMESTAMP_LTZ(create_time, 3),
— 定义 Watermark：允许 5 秒乱序
WATERMARK FOR ts AS ts – INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'project-order-topic',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'sql-platform-group',
'scan.startup.mode' = 'latest-offset',
'format' = 'json',
'json.fail-on-missing-field' = 'false',
'json.ignore-parse-errors' = 'true'
);

这张 DDL 蕴含了实时计算平台中最常用的三类声明式能力：

  • 计算列与事件时间提取：ts AS TO_TIMESTAMP_LTZ(create_time, 3) 是一个计算列，把 BIGINT 毫秒时间戳转换成 TIMESTAMP_LTZ(3) 类型的事件时间属性，其中 3 表示精度到毫秒；
  • Watermark 策略：WATERMARK FOR ts AS ts – INTERVAL '5' SECOND 声明允许 5 秒的乱序/延迟数据，超过该延迟的迟到数据会被丢弃或进入侧输出，直接决定窗口触发的准确性与实时性之间的平衡；
  • Kafka Connector DDL 参数：scan.startup.mode = 'latest-offset' 表示作业启动时从最新位点消费（生产环境可改为 earliest-offset 或指定 timestamp/offsets 位点）；json.fail-on-missing-field = 'false' 与 json.ignore-parse-errors = 'true' 增强了脏数据的容忍度，避免单条坏数据拖垮整个作业。

    • 2.3 用 DDL 定义 Print Sink 表（调试输出）

    CREATE TABLE print_sink (
    category STRING,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    order_count BIGINT,
    total_amount DOUBLE,
    unique_users BIGINT
    ) WITH (
    'connector' = 'print'
    );

    print 连接器会把每条结果打印到 TaskManager 的标准输出，适合平台开发阶段的联调验证；生产环境应替换为 Kafka Sink（'connector' = 'kafka'）或 JDBC/ClickHouse 等存储连接器。

    2.4 核心聚合 SQL：TUMBLE 窗口 + COUNT DISTINCT UV 去重

    INSERT INTO print_sink
    SELECT
    category,
    TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start,
    TUMBLE_END(ts, INTERVAL '5' MINUTE) AS window_end,
    COUNT(*) AS order_count,
    SUM(total_amount) AS total_amount,
    COUNT(DISTINCT user_id) AS unique_users
    FROM ods_order
    WHERE total_amount > 0
    GROUP BY category, TUMBLE(ts, INTERVAL '5' MINUTE);

    这段 SQL 汇聚了实时平台中三个高频知识点：

    • TUMBLE 滚动窗口：TUMBLE(ts, INTERVAL '5' MINUTE) 按事件时间 ts 每 5 分钟切一个窗口，TUMBLE_START/TUMBLE_END 取出窗口起止时间；
    • 多维窗口聚合：GROUP BY category, TUMBLE(…) 在按类别分桶的基础上再开窗口，输出每个类别每个窗口的统计结果；
    • UV 去重：COUNT(DISTINCT user_id) 在窗口内统计去重用户数，等价于 UV 指标；COUNT(*) 与 SUM(total_amount) 则分别对应 PV/GMV 类指标。

    执行 tableEnv.executeSql(aggregationSQL) 后返回 TableResult。源码注释明确指出：INSERT INTO 是异步执行的，流模式下作业会持续运行直到被取消，这是实时平台「常驻任务」的典型特征。

    三、模式二：TableApiExampleJob——Table API 编程构建实时计算任务

    TableApiExampleJob 以「页面访问行为分析」为业务场景，展示编程方式下的完整链路：KafkaSource 建流 → DataStream 转 Table → 编程式查询 → 动态表转回 DataStream。

    3.1 从 Kafka 消费数据并生成带 Watermark 的 DataStream

    KafkaSource<String> kafkaSource = ProjectKafkaUtil.buildKafkaStringSource(
    ProjectConstants.TOPIC_PAGE_ACCESS, "table-api-example-group");

    WatermarkStrategy<PageAccessEvent> watermarkStrategy = WatermarkStrategy
    .<PageAccessEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner(
    (SerializableTimestampAssigner<PageAccessEvent>) (event, ts) ->
    event.getTimestamp());

    DataStream<PageAccessEvent> accessStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Page Access Source")
    .map(json -> GsonUtil.fromJson(json, PageAccessEvent.class))
    .assignTimestampsAndWatermarks(watermarkStrategy);

    源码细节印证：

    • Kafka 连接器使用新版 KafkaSource API，具体构造逻辑封装在 ProjectKafkaUtil.java 的 buildKafkaStringSource 中：SimpleStringSchema 反序列化、OffsetsInitializer.latest() 最新位点起步、默认 broker 地址取 ProjectConstants.java 中的 localhost:9092；
    • 消息先以 JSON 字符串形式消费，再通过 GsonUtil.fromJson 反序列化为 PageAccessEvent.java（含 userId、pageCategory、channel、stayDuration、timestamp、region 等字段，注释表明该模型服务于实时大屏 PV/UV 统计）；
    • Watermark 使用 forBoundedOutOfOrderness(Duration.ofSeconds(5)) 允许 5 秒乱序，这与 SQL 方式中 INTERVAL '5' SECOND 的乱序容忍度保持一致，说明两种模式在语义上是等价的。

    3.2 DataStream → Table：fromDataStream 字段映射

    Table accessTable = tableEnv.fromDataStream(accessStream,
    $("eventId"),
    $("userId"),
    $("pageUrl"),
    $("pageTitle"),
    $("pageCategory"),
    $("channel"),
    $("stayDuration"),
    $("timestamp").as("ts"), // 重命名事件时间字段
    $("deviceType"),
    $("region"));

    fromDataStream 把带类型信息的 DataStream 转换为 Table，并显式声明字段顺序；$("timestamp").as("ts") 是 Expression API 的字段重命名，为后续窗口聚合准备好事件时间字段 ts。

    3.3 编程式查询：filter + select + groupBy + 聚合

    Table filteredTable = accessTable
    .filter($("channel").isNotNull())
    .select($("pageCategory"), $("channel"), $("userId"), $("stayDuration"), $("region"));

    Table channelStats = filteredTable
    .groupBy($("channel"))
    .select(
    $("channel"),
    $("userId").count().as("pv"),
    $("stayDuration").avg().as("avg_stay")
    );

    • filter($("channel").isNotNull()) 过滤掉渠道为空的数据；
    • $("userId").count() 统计各渠道访问次数（近似 PV），$("stayDuration").avg() 计算平均停留时长；
    • 整条查询全部通过静态导入的 $ / lit 表达式构建，这正是 README 中「Expression API：$("field") 字段引用」知识点的落地。

    3.4 Table → DataStream：toChangelogStream 输出结果

    DataStream<Row> resultStream = tableEnv.toChangelogStream(channelStats);
    resultStream.print("channel-stats");
    env.execute("Table API 编程示例");

    由于 channelStats 是带更新的聚合结果（动态表），转回 DataStream 必须使用 toChangelogStream（而非 toAppendStream），得到的是包含 +I/-U/+U 等变更类型的 Changelog 流，打印输出即为流式更新语义下的渠道统计结果。这一步也是 README 中「Changelog 流：动态表转回 DataStream」的代码级体现。

    四、Flink SQL vs Table API：选型对比

    README 给出了平台侧最关心的对比维度，结合两个作业的源码实现可以进一步明确各自的适用边界：

    特性Flink SQLTable API
    使用方式 纯 SQL 字符串 Java/Scala 编程
    适用人群 数据分析师 平台开发者
    灵活性 中等 高
    可调试性 低 高
    动态提交 容易 较难

    从实现上看：

    • SQL 方式（FlinkSqlPlatformJob）：建表、窗口、聚合全部是字符串，天然适合做成「SQL 即任务」的平台化提交——平台接收用户 SQL 后动态拼接执行即可，无需重新编译打包，这是实时计算平台最常见的形态；
    • Table API 方式（TableApiExampleJob）：逻辑以代码表达，可与 DataStream API 自由互转（fromDataStream/toChangelogStream），便于复用 ProjectKafkaUtil、GsonUtil 等既有工具链，适合业务逻辑复杂、需要精细控制（如自定义 Watermark 分配、旁路输出）的开发场景；
    • 两者的底层执行引擎与优化器是同一套，区别只在「表达层」，因此同一业务用两种方式书写，执行语义一致。

    五、涉及的核心 Flink 知识点速览

    README 中的知识点表与上述两个作业一一对应，汇总如下：

    知识点说明所在类
    StreamTableEnvironment 流表桥接环境 全部
    SQL DDL 用 SQL 定义表结构和连接器 FlinkSqlPlatformJob
    Kafka Connector DDL SQL 中配置 Kafka 连接器 FlinkSqlPlatformJob
    TUMBLE Window SQL 滚动窗口函数 FlinkSqlPlatformJob
    COUNT DISTINCT SQL UV 去重 FlinkSqlPlatformJob
    fromDataStream DataStream 转 Table TableApiExampleJob
    toChangelogStream Table 转 DataStream TableApiExampleJob
    Expression API $("field") 字段引用 TableApiExampleJob

    六、如何运行本模块

    仓库为多模块 Maven 工程，模块间存在依赖关系（本模块依赖 flink-learning-project-common），建议在仓库根目录先构建共享模块，再运行目标作业：

    # 在仓库根目录下编译项目公共模块
    mvn -pl flink-learning-project/flink-learning-project-common -am install -DskipTests
    # 编译实时计算平台模块
    mvn -pl flink-learning-project/flink-learning-project-real-time-computing-platform -am package -DskipTests

    运行前提：

  • 本地或集群存在 Kafka，且 project-order-topic、project-page-access-topic 两个 topic 可用（topic 名称定义于 ProjectConstants.java）；
  • localhost:9092 为默认 broker 地址，按需修改；
  • FlinkSqlPlatformJob 需要持续向 project-order-topic 生产符合 DDL 结构的 JSON 数据，才能观察到窗口聚合结果；TableApiExampleJob 同理需要页面访问事件数据，其 JSON 字段与 PageAccessEvent.java 的字段一一对应。

    • 七、小结：从示例到平台化落地

    本模块的价值不在于单个作业，而在于示范了实时计算平台的两条核心能力路径：SQL 化动态提交（DDL + INSERT INTO，支撑分析师自助取数与指标计算）与 编程化任务开发（DataStream ↔ Table 互转，支撑平台开发者实现复杂逻辑）。结合仓库中 flink-learning-project-real-time-data-warehouse、flink-learning-project-real-time-computing-platform 等兄弟模块可以看到，这类 SQL/Table 双模式模板正是整个项目体系中「数仓分层、指标计算、平台建设」公共底座的一部分。开发者可在此基础上，将示例中的 Print Sink 替换为真实存储连接器、将常量提取到配置中心，即可快速演化出可上线的实时计算平台作业。

    赞

    分享

    • 示例工程
    • 大数据

    【免费下载链接】flink-learning

    flink learning blog. http://www.54tianzhisheng.cn/ 含 Flink 入门、概念、原理、实战、性能调优、源码解析等内容。涉及 Flink Connector、Metrics、Library、DataStream API、Table API & SQL 等内容的学习案例，还有 Flink 落地应用的大型项目案例（PVUV、日志存储、百亿数据实时去重、监控告警）分享。欢迎大家支持我的专栏《大数据实时计算引擎 Flink 实战与性能优化》

    项目地址：
    https://gitcode.com/gh_mirrors/fl/flink-learning

    点击查看 免费下载

    相关推荐

    • Flink性能对决：Table API vs SQL，谁才是实时计算王者？

      Flink性能对决：Table API vs SQL，谁才是实时计算王者？ 你还在为实时数据处理选择API烦恼？Table API的灵活编码与SQL的简洁高效，

      后端
      大数据
      流处理
      批处理

    • Dinky实时计算平台：让Flink SQL开发更高效

      Dinky实时计算平台：让Flink SQL开发更高效 什么是Dinky Dinky是一款基于Apache Flink构建的一站式实时计算平台，其核心理念是"实

      大数据
      后端
      前端
      实时分析
      流处理
      数据集成

    • 革命性日语文本生成模型：jeffding/japanese-gpt2-small-openmind完全解析

      革命性日语文本生成模型：jeffding/japanese gpt2 small openmind完全解析 jeffding/japanese gpt2 sma

    上一篇：
    深度解析Fay AI Agent框架：构建智能数字人系统的技术实现方案

    下一篇：
    Node.js 生产环境最佳实践：正确设置 NODE_ENV=production 提升性能

    创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

    赞(0)
    未经允许不得转载：171主机测评 » 基于 Apache Flink 的实时计算平台：Flink SQL 与 Table API 双模式实战指南
    分享到： 更多 (0)

    相关推荐

    评论 抢沙发

    取消

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址

    热门标签

    #人工智能(64070)#python(21377)#开发语言(17787)#java(16364)#运维(16133)#linux(15615)#大数据(15288)#服务器(13633)#算法(12269)#前端(11620)#ai(10200)#网络(9882)#数据库(9477)#c++(8625)#学习(7153)#课程设计(7028)#机器学习(6996)#区块链(6051)#后端(5520)#OpenClaw(5412)#spring boot(5281)#深度学习(5259)#c语言(4546)#安全(4476)#毕业设计(4395)#数据结构(4336)#架构(4159)#docker(4108)#javascript(3970)#大模型(3901)

    热门文章

    最新文章

    归档

    分类