Flink 实践教程 - 进阶（11）：SQL 关联：Regular Join

2022 年 4 月 01 日
本文字数：2588 字
阅读完需：约 8 分钟

作者：腾讯云流计算 Oceanus 团队

流计算 Oceanus 简介

流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。

Flink SQL 提供了 Regular Joins、Interval Joins、Temporal Joins、Lookup Join、Array 展平和 Table Function 六种方式实现数据关联。

本文将为您介绍如何使用 Regualr Joins 实现数据关联。Regualr Joins 在使用时有一定的限制条件，比如只能在 Equi-Join 条件下使用。下面将以 Kafka 作为源表的左右表为例，将商品订单 order-source 中商品 ID 与 product-info 中商品 ID 进行左关联得到商品名称，最终将结果数据到 Logger Sink 中去。

教程链接：https://cloud.tencent.com/developer/video/30595?sharedUid=1077144

前置准备

创建流计算 Oceanus 集群

进入 Oceanus 控制台 [1]，点击左侧【集群管理】，点击左上方【创建集群】，具体可参考 Oceanus 官方文档 创建独享集群 [2]。

创建 Kafka Topic

进入 CKafka 控制台 [3]，点击左上角【新建】，即可完成 CKafka 实例的创建，并创建 2 个 Topic，order-source 和 product-info 。

流计算 Oceanus 作业

1. 上传依赖

在 Oceanus 控制台，点击左侧【依赖管理】，点击左上角【新建】新建依赖，上传 Logger Sink[4] JAR 包。

2. 创建作业

在 Oceanus 控制台，点击左侧【作业管理】，点击左上角【新建】新建作业，作业类型选择 SQL 作业，点击【开发调试】进入作业编辑页面。

CREATE TABLE `order_source` (   `id` INT,   `user_id` INT,   `product_id` INT,   `create_time` TIMESTAMP(3)) WITH ( 'connector' = 'kafka', 'topic' = 'order-source',   'scan.startup.mode' = 'latest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = 'x.x.x.x:9092',  -- 替换为您的 Kafka 连接地址 'properties.group.id' = 'testGroup0', -- 必选参数, 一定要指定 Group ID 'format' = 'json', 'json.fail-on-missing-field' = 'false',  -- 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true'    -- 如果设置为 true，则忽略任何解析报错。);CREATE TABLE `product_info` (   `product_id` INT,   `product_name` STRING,   `update_time` TIMESTAMP(3)) WITH ( 'connector' = 'kafka', 'topic' = 'product-info',  -- 替换为您要消费的 Topic 'scan.startup.mode' = 'latest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种 'properties.bootstrap.servers' = 'x.x.x.x:9092',  -- 替换为您的 Kafka 连接地址 'properties.group.id' = 'testGroup0', -- 必选参数, 一定要指定 Group ID 'format' = 'json', 'json.fail-on-missing-field' = 'false',  -- 如果设置为 false, 则遇到缺失字段不会报错。 'json.ignore-parse-errors' = 'true'    -- 如果设置为 true，则忽略任何解析报错。);CREATE TABLE logger_sink_table (   `id` INT PRIMARY KEY NOT ENFORCED,   `user_id` INT,   `product_id` INT,   `product_name` STRING,   `create_time` TIMESTAMP(3)) WITH (   'connector' = 'logger',   'print-identifier' = 'DebugData');INSERT INTO logger_sink_tableSELECT order_source.id, order_source.user_id, order_source.product_id, product_info.product_name, order_source.create_timeFROM order_source left join product_infoon order_source.product_id = product_info.product_id;

复制代码

3. 运行作业

点击【发布草稿】后启动作业，可通过【日志】面板 TaskManager 或 Flink UI 查看运行信息。

4. 模拟数据

通过 Kafka Client 发送数据到关联的左表 order-source 和右表 product-info。发送消息命令：

[root@VM-3-centos ~]# cd /opt/kafka_2.11-2.4.1/bin/[root@VM-3-centos bin]# bash kafka-console-producer.sh --broker-list 10.0.0.29:9092 --topic order-source

复制代码

Topic order-source 模拟数据示例：

{"id":1,"user_id":10,"product_id":1001,"create_time":"2022-03-17 16:47:00"}{"id":2,"user_id":10,"product_id":1001,"create_time":"2022-03-17 16:48:00"}{"id":3,"user_id":10,"product_id":1002,"create_time":"2022-03-17 16:49:00"}

复制代码

发送消息命令：

[root@VM-3-centos ~]# cd /opt/kafka_2.11-2.4.1/bin/[root@VM-3-centos bin]# bash kafka-console-producer.sh --broker-list 10.0.0.29:9092 --topic product-info

复制代码

Topic product-info 模拟数据示例：

{"id":1,"user_id":10,"product_id":1001,"create_time":"2022-03-17 16:47:00"}{"id":2,"user_id":10,"product_id":1001,"create_time":"2022-03-17 16:48:00"}{"id":3,"user_id":10,"product_id":1002,"create_time":"2022-03-17 16:49:00"}

复制代码