devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

FlinkCDC Java DataStream API 的同源双流JOIN

作者：笔触狂放9 | 2024-07-03 10:26:38

赞

踩

flink datastream api实现cdc

7e3dbf8576c9d4a82dedefbea607d037.png全网最全大数据面试提升手册！

1 说明

1.1 案例说明

本文使用 Flink CDC 最新版本 2.2 及 Flink 1.14 版本通过 Java DataStream API 做 双流 Join 案例。

双流 Join 大致流程：
d32470bf04c12fae137d5a438f6b3eb3.jpeg
双流Join案例具体划分有：
时间窗口联结

  • 处理事件窗口联结

    • 处理时间窗口 内联结 案例

    • 处理时间窗口 外联结 案例

  • 事件时间窗口联结

    • 事件时间窗口 内联结 案例

    • 事件时间窗口 外联结 案例

间隔联结

案例大致流程如下：

03ed5f7baa959be82a97ed2d901500ef.jpeg
1.2 软件说明

软件版本说明如下：

63dcc07e20c2fed8c060f64a9c7cf862.png

注意事项：

  • MySQL 服务器必须开启 binlog 功能

  • Flink CDC 官网：https://github.com/ververica/flink-cdc-connectors

  • Flink 官网：https://flink.apache.org/

1.3 数据表字段说明

教师表（teacher）字段说明：

  1. [
  2.     t_id varchar(3) primary key COMMENT '主键',
  3.     t_name varchar(10) not null COMMENT '主键',
  4. ]

课程表（course）字段说明：

  1. [
  2.     c_id varchar(3) primary key COMMENT '主键',
  3.     c_name varchar(20) not null COMMENT '主键',
  4.     c_tid varchar(3) not null COMMENT '主键'
  5. ]
1.4 数据流字段说明

教师表（teacher）数据流字段说明：

  1. [
  2.   t_id:  string,   // 主键id
  3.   t_name: string, // 教师名称
  4.   op: string, // 数据操作类型
  5.   ts_ms: long // 毫秒时间戳
  6. ]

课程表（course）数据流字段说明：

  1. [
  2.   c_id:  string,   // 主键id
  3.   c_name: string, // 课程名称
  4.   c_tid: string,  // 教师表主键id
  5.   op: string, // 数据操作类型
  6.   ts_ms: long // 毫秒时间戳
  7. ]

2 准备工作

2.1 MySQL 数据准备

用 Navicat 连接 MySQL，建表及初始化数据相关 SQL 如下：

  1. # 创建数据库 flinkcdc_etl_test
  2. create database flinkcdc_etl_test;
  3.  
  4. # 使用数据库 flinkcdc_etl_test
  5. use flinkcdc_etl_test;
  6.  
  7. # 创建教师表
  8. -- ----------------------------
  9. -- Table structure for teacher
  10. -- ----------------------------
  11. DROP TABLE IF EXISTS `teacher`;
  12. CREATE TABLE `teacher`  (
  13.   `t_id` varchar(3) NOT NULL COMMENT '主键',
  14.   `t_name` varchar(10)  NOT NULL COMMENT '教师名称',
  15.   PRIMARY KEY (`t_id`) USING BTREE
  16. )COMMENT = '教师表';
  17.  
  18. -- ----------------------------
  19. -- Records of teacher
  20. -- ----------------------------
  21. INSERT INTO `teacher` VALUES ('001', '张三');
  22. INSERT INTO `teacher` VALUES ('002', '李四');
  23. INSERT INTO `teacher` VALUES ('003', '王五');
  24.  
  25. -- ----------------------------
  26. -- Table structure for course
  27. -- ----------------------------
  28. DROP TABLE IF EXISTS `course`;
  29. CREATE TABLE `course`  (
  30.   `c_id` varchar(3)  NOT NULL COMMENT '主键',
  31.   `c_name` varchar(20)  NOT NULL COMMENT '课程名称',
  32.   `c_tid` varchar(3)  NOT NULL COMMENT '教师表主键',
  33.   PRIMARY KEY (`c_id`) USING BTREE,
  34.   INDEX `c_tid`(`c_tid`) USING BTREE
  35. )COMMENT = '课程表';
  36.  
  37. -- ----------------------------
  38. -- Records of course
  39. -- ----------------------------
  40. INSERT INTO `course` VALUES ('1', '语文', '001');
  41. INSERT INTO `course` VALUES ('2', '数学', '002');

操作截图如下：

350abdcf702a5018f1bb244eb96027df.jpeg
2.2 项目工程
2.2.1 项目结构

在 IDEA 中创建 maven 项目，项目结构截图如下：

9a2372844037a390a4c3f42921976db6.jpeg
2.2.2 项目依赖

pom.xml 文件内容如下：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.  
  7.     <groupId>cn.mfox</groupId>
  8.     <artifactId>teacher-course-etl-demo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.  
  11.     <properties>
  12.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  13.         <maven.compiler.source>1.8</maven.compiler.source>
  14.         <maven.compiler.target>1.8</maven.compiler.target>
  15.         <flink.version>1.14.4</flink.version>
  16.         <scala.binary.version>2.12</scala.binary.version>
  17.     </properties>
  18.  
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.apache.flink</groupId>
  22.             <artifactId>flink-streaming-java_2.12</artifactId>
  23.             <version>${flink.version}</version>
  24.         </dependency>
  25.         <dependency>
  26.             <groupId>org.apache.flink</groupId>
  27.             <artifactId>flink-java</artifactId>
  28.             <version>${flink.version}</version>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>org.apache.flink</groupId>
  32.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
  33.             <version>${flink.version}</version>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>mysql</groupId>
  37.             <artifactId>mysql-connector-java</artifactId>
  38.             <version>8.0.23</version>
  39.         </dependency>
  40.  
  41.         <dependency>
  42.             <groupId>com.ververica</groupId>
  43.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
  44.             <version>2.2.0</version>
  45.         </dependency>
  46.  
  47.         <dependency>
  48.             <groupId>org.apache.flink</groupId>
  49.             <artifactId>flink-table-planner-blink_2.12</artifactId>
  50.             <version>1.12.0</version>
  51.         </dependency>
  52.  
  53.         <dependency>
  54.             <groupId>com.alibaba</groupId>
  55.             <artifactId>fastjson</artifactId>
  56.             <version>1.2.80</version>
  57.         </dependency>
  58.     </dependencies>
  59.     <build>
  60.         <plugins>
  61.             <plugin>
  62.                 <artifactId>maven-compiler-plugin</artifactId>
  63.                 <version>3.6.0</version>
  64.                 <configuration>
  65.                     <source>1.8</source>
  66.                     <target>1.8</target>
  67.                 </configuration>
  68.             </plugin>
  69.         </plugins>
  70.     </build>
  71. </project>
2.2.3 基础类

OpEnum.java 枚举类内容如下：

  1. package cn.mfox.enumeration;
  2.  
  3. /**
  4.  * CDC 中op类型
  5.  *
  6.  * @author hy
  7.  * @version 1.0
  8.  * @date 2022/4/08 17:28
  9.  */
  10. public enum OpEnum {
  11.  
  12.     /**
  13.      * 新增
  14.      */
  15.     CREATE("c", "create", "新增"),
  16.  
  17.     /**
  18.      * 修改
  19.      */
  20.     UPDATA("u", "update", "更新"),
  21.  
  22.     /**
  23.      * 删除
  24.      */
  25.     DELETE("d", "delete", "删除"),
  26.  
  27.     /**
  28.      * 读
  29.      */
  30.     READ("r", "read", "首次读");
  31.  
  32.     /**
  33.      * 字典码
  34.      */
  35.     private String dictCode;
  36.  
  37.     /**
  38.      * 字典码翻译值
  39.      */
  40.     private String dictValue;
  41.  
  42.     /**
  43.      * 字典码描述
  44.      */
  45.     private String description;
  46.  
  47.     OpEnum(String dictCode, String dictValue, String description) {
  48.         this.dictCode = dictCode;
  49.         this.dictValue = dictValue;
  50.         this.description = description;
  51.     }
  52.  
  53.     public String getDictCode() {
  54.         return dictCode;
  55.     }
  56.  
  57.     public String getDictValue() {
  58.         return dictValue;
  59.     }
  60.  
  61.     public String getDescription() {
  62.         return description;
  63.     }
  64. }

TransformUtil.java 工具类内容如下：

  1. package cn.mfox.utils;
  2.  
  3. import cn.mfox.enumeration.OpEnum;
  4. import com.alibaba.fastjson.JSONObject;
  5.  
  6. /**
  7.  * 转换工具类
  8.  *
  9.  * @author hy
  10.  * @version 1.0
  11.  * @date 2022/5/6 16:25
  12.  */
  13. public class TransformUtil {
  14.  
  15.     /**
  16.      * 格式化抽取数据格式
  17.      * 去除before、after、source等冗余内容
  18.      *
  19.      * @param extractData 抽取的数据
  20.      * @return
  21.      */
  22.     public static JSONObject formatResult(String extractData) {
  23.         JSONObject formatDataObj = new JSONObject();
  24.         JSONObject rawDataObj = JSONObject.parseObject(extractData);
  25.         formatDataObj.putAll(rawDataObj);
  26.         formatDataObj.remove("before");
  27.         formatDataObj.remove("after");
  28.         formatDataObj.remove("source");
  29.         String op = rawDataObj.getString("op");
  30.         if (OpEnum.DELETE.getDictCode().equals(op)) {
  31.             // 新增取 before结构体数据
  32.             formatDataObj.putAll(rawDataObj.getJSONObject("before"));
  33.         } else {
  34.             // 其余取 after结构体数据
  35.             formatDataObj.putAll(rawDataObj.getJSONObject("after"));
  36.         }
  37.         return formatDataObj;
  38.     }
  39. }
2.2.4 水位线类

水位线类截图如下：

29e7245a28f066dc76061dc7281480d0.jpeg

CourseDataStreamNoWatermark.java 内容如下：

  1. package cn.mfox.etl.v2.join.watermark;
  2.  
  3. import cn.mfox.utils.TransformUtil;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  6. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  7. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  8. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12.  
  13. /**
  14.  * 抽取 课程表表数据源，无水位线
  15.  *
  16.  * @author hy
  17.  * @version 1.0
  18.  * @date 2022/5/6 16:33
  19.  */
  20. public class CourseDataStreamNoWatermark {
  21.  
  22.     /**
  23.      * 获取 课程表 数据流
  24.      *
  25.      * @param env
  26.      * @return
  27.      */
  28.     public static DataStream<JSONObject> getCourseDataStream(StreamExecutionEnvironment env) {
  29.         // 1.创建Flink-MySQL-CDC的Source
  30.         MySqlSource<String> courseSouce = MySqlSource.<String>builder()
  31.                 .hostname("192.168.18.101")
  32.                 .port(3306)
  33.                 .username("root")
  34.                 .password("123456")
  35.                 .databaseList("flinkcdc_etl_test")
  36.                 .tableList("flinkcdc_etl_test.course")
  37.                 .startupOptions(StartupOptions.initial())
  38.                 .deserializer(new JsonDebeziumDeserializationSchema())
  39.                 .serverTimeZone("Asia/Shanghai")
  40.                 .build();
  41.  
  42.         // 2.使用CDC Source从MySQL读取数据
  43.         DataStreamSource<String> mysqlDataStreamSource = env.fromSource(
  44.                 courseSouce,
  45.                 WatermarkStrategy.noWatermarks(),
  46.                 "CourseDataStreamNoWatermark Source"
  47.         );
  48.         // 3.转换为指定格式
  49.         DataStream<JSONObject> courseDataStream = mysqlDataStreamSource.map(rawData -> {
  50.             return TransformUtil.formatResult(rawData);
  51.         });
  52.         return courseDataStream;
  53.     }
  54. }

CourseDataStreamWithWatermark.java 内容如下：

  1. package cn.mfox.etl.v2.join.watermark;
  2.  
  3. import cn.mfox.utils.TransformUtil;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  6. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  7. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  8. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  9. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  10. import org.apache.flink.streaming.api.datastream.DataStream;
  11. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13.  
  14. import java.time.Duration;
  15.  
  16. /**
  17.  * 抽取 课程表表数据源
  18.  * 有水位线，指定ts_ms为时间戳
  19.  *
  20.  * @author hy
  21.  * @version 1.0
  22.  * @date 2022/5/6 16:33
  23.  */
  24. public class CourseDataStreamWithWatermark {
  25.  
  26.     /**
  27.      * 获取 课程表 数据流
  28.      *
  29.      * @param env
  30.      * @return
  31.      */
  32.     public static DataStream<JSONObject> getCourseDataStream(StreamExecutionEnvironment env) {
  33.         // 1.创建Flink-MySQL-CDC的Source
  34.         MySqlSource<String> courseSouce = MySqlSource.<String>builder()
  35.                 .hostname("192.168.18.101")
  36.                 .port(3306)
  37.                 .username("root")
  38.                 .password("123456")
  39.                 .databaseList("flinkcdc_etl_test")
  40.                 .tableList("flinkcdc_etl_test.course")
  41.                 .startupOptions(StartupOptions.initial())
  42.                 .deserializer(new JsonDebeziumDeserializationSchema())
  43.                 .serverTimeZone("Asia/Shanghai")
  44.                 .build();
  45.  
  46.         // 2.使用CDC Source从MySQL读取数据
  47.         DataStreamSource<String> mysqlDataStreamSource = env.fromSource(
  48.                 courseSouce,
  49.                 WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(1L)).withTimestampAssigner(
  50.                         new SerializableTimestampAssigner<String>() {
  51.                             @Override
  52.                             public long extractTimestamp(String extractData, long l) {
  53.                                 return JSONObject.parseObject(extractData).getLong("ts_ms");
  54.                             }
  55.                         }
  56.                 ),
  57.                 "CourseDataStreamWithWatermark Source"
  58.         );
  59.         // 3.转换为指定格式
  60.         DataStream<JSONObject> courseDataStream = mysqlDataStreamSource.map(rawData -> {
  61.             return TransformUtil.formatResult(rawData);
  62.         });
  63.         return courseDataStream;
  64.     }
  65. }

TeacherDataStreamNoWatermark.java 内容如下：

  1. package cn.mfox.etl.v2.join.watermark;
  2.  
  3. import cn.mfox.utils.TransformUtil;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  6. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  7. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  8. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12.  
  13. /**
  14.  * 抽取 教师表数据源
  15.  * 无水位线
  16.  *
  17.  * @author hy
  18.  * @version 1.0
  19.  * @date 2022/5/6 16:33
  20.  */
  21. public class TeacherDataStreamNoWatermark {
  22.  
  23.     /**
  24.      * 获取教师表数据流
  25.      *
  26.      * @param env
  27.      * @return
  28.      */
  29.     public static DataStream<JSONObject> getTeacherDataStream(StreamExecutionEnvironment env) {
  30.         // 1.创建Flink-MySQL-CDC的Source
  31.         MySqlSource<String> teacherSouce = MySqlSource.<String>builder()
  32.                 .hostname("192.168.18.101")
  33.                 .port(3306)
  34.                 .username("root")
  35.                 .password("123456")
  36.                 .databaseList("flinkcdc_etl_test")
  37.                 .tableList("flinkcdc_etl_test.teacher")
  38.                 .startupOptions(StartupOptions.initial())
  39.                 .deserializer(new JsonDebeziumDeserializationSchema())
  40.                 .serverTimeZone("Asia/Shanghai")
  41.                 .build();
  42.  
  43.         // 2.使用CDC Source从MySQL读取数据
  44.         DataStreamSource<String> mysqlDataStreamSource = env.fromSource(
  45.                 teacherSouce,
  46.                 WatermarkStrategy.noWatermarks(),
  47.                 "TeacherDataStreamNoWatermark Source"
  48.         );
  49.         // 3.转换为指定格式
  50.         DataStream<JSONObject> teacherDataStream = mysqlDataStreamSource.map(rawData -> {
  51.             return TransformUtil.formatResult(rawData);
  52.         });
  53.         return teacherDataStream;
  54.     }
  55.  
  56.     public static void main(String[] args) {
  57.         System.out.println(System.currentTimeMillis());
  58.     }
  59. }

TeacherDataStreamWithWatermark.java 内容如下：

  1. package cn.mfox.etl.v2.join.watermark;
  2.  
  3. import cn.mfox.utils.TransformUtil;
  4. import com.alibaba.fastjson.JSONObject;
  5. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  6. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  7. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  8. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  9. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  10. import org.apache.flink.streaming.api.datastream.DataStream;
  11. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13.  
  14. import java.time.Duration;
  15.  
  16. /**
  17.  * 抽取 教师表数据源
  18.  * 有水位线，指定ts_ms为时间戳
  19.  *
  20.  * @author hy
  21.  * @version 1.0
  22.  * @date 2022/5/6 16:33
  23.  */
  24. public class TeacherDataStreamWithWatermark {
  25.  
  26.     /**
  27.      * 获取教师表数据流
  28.      *
  29.      * @param env
  30.      * @return
  31.      */
  32.     public static DataStream<JSONObject> getTeacherDataStream(StreamExecutionEnvironment env) {
  33.         // 1.创建Flink-MySQL-CDC的Source
  34.         MySqlSource<String> teacherSouce = MySqlSource.<String>builder()
  35.                 .hostname("192.168.18.101")
  36.                 .port(3306)
  37.                 .username("root")
  38.                 .password("123456")
  39.                 .databaseList("flinkcdc_etl_test")
  40.                 .tableList("flinkcdc_etl_test.teacher")
  41.                 .startupOptions(StartupOptions.initial())
  42.                 .deserializer(new JsonDebeziumDeserializationSchema())
  43.                 .serverTimeZone("Asia/Shanghai")
  44.                 .build();
  45.  
  46.         // 2.使用CDC Source从MySQL读取数据
  47.         DataStreamSource<String> mysqlDataStreamSource = env.fromSource(
  48.                 teacherSouce,
  49.                 WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(1L)).withTimestampAssigner(
  50.                         new SerializableTimestampAssigner<String>() {
  51.                             @Override
  52.                             public long extractTimestamp(String extractData, long l) {
  53.                                 return JSONObject.parseObject(extractData).getLong("ts_ms");
  54.                             }
  55.                         }
  56.                 ),
  57.                 "TeacherDataStreamWithWatermark Source"
  58.         );
  59.         // 3.转换为指定格式
  60.         DataStream<JSONObject> teacherDataStream = mysqlDataStreamSource.map(rawData -> {
  61.             return TransformUtil.formatResult(rawData);
  62.         });
  63.         return teacherDataStream;
  64.     }
  65. }

3 时间窗口联结

3.1 处理时间窗口
3.1.1 处理时间窗口内联结

WindowInnerJoinByProcessTimeTest.java 内容如下：

  1. package cn.mfox.etl.v2.join.window.inner;
  2.  
  3. import cn.mfox.etl.v2.join.watermark.CourseDataStreamNoWatermark;
  4. import cn.mfox.etl.v2.join.watermark.TeacherDataStreamNoWatermark;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flink.api.common.functions.JoinFunction;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11.  
  12. /**
  13.  * 基于 处理时间 的 window inner join 把教师表和课程表进行联结
  14.  * <p>
  15.  * 只有两者数据流关联到数据，才会进行打印
  16.  *
  17.  * @author hy
  18.  * @version 1.0
  19.  * @date 2022/5/6 16:31
  20.  */
  21. public class WindowInnerJoinByProcessTimeTest {
  22.  
  23.     public static void main(String[] args) throws Exception {
  24.         // 1.创建执行环境
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         // 2.获取教师数据流和课程表数据流
  28.         DataStream<JSONObject> teacherDataStream = TeacherDataStreamNoWatermark.getTeacherDataStream(env);
  29.         DataStream<JSONObject> courseDataStream = CourseDataStreamNoWatermark.getCourseDataStream(env);
  30.         // 3.窗口联结（教师流和课程表）打印输出
  31.         windowInnerJoinAndPrint(teacherDataStream, courseDataStream);
  32.         // 4.执行任务
  33.         env.execute("WindowInnerJoinByProcessTimeTest Job");
  34.     }
  35.  
  36.     /**
  37.      * 窗口联结并打印输出
  38.      * 只支持 inner join，即窗口内联关联到的才会下发，关联不到的则直接丢掉。
  39.      * 如果想实现Window上的 outer join，需要使用coGroup算子
  40.      *
  41.      * @param teacherDataStream 教师数据流
  42.      * @param courseDataStream  课程数据流
  43.      */
  44.     private static void windowInnerJoinAndPrint(DataStream<JSONObject> teacherDataStream,
  45.                                                 DataStream<JSONObject> courseDataStream) {
  46.         DataStream<JSONObject> teacherCourseDataStream = teacherDataStream
  47.                 .join(courseDataStream)
  48.                 .where(teacher -> teacher.getString("t_id"))
  49.                 .equalTo(couse -> couse.getString("c_tid"))
  50.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(3L)))
  51.                 .apply(
  52.                         new JoinFunction<JSONObject, JSONObject, JSONObject>() {
  53.                             @Override
  54.                             public JSONObject join(JSONObject jsonObject,
  55.                                                    JSONObject jsonObject2) {
  56.                                 // 拼接
  57.                                 jsonObject.putAll(jsonObject2);
  58.                                 return jsonObject;
  59.                             }
  60.                         }
  61.                 );
  62.         teacherCourseDataStream.print("Window Inner Join By Process Time");
  63.     }
  64. }

程序运行截图：

3c8bdd570af5d39ccec3f3af148c500c.jpeg

控制台输出内容：

  1. Window Inner Join By Process Time> {"t_id":"001","op":"r","c_tid":"001","t_name":"张三","c_id":"1","c_name":"语文","ts_ms":1652259235842}
  2. Window Inner Join By Process Time> {"t_id":"002","op":"r","c_tid":"002","t_name":"李四","c_id":"2","c_name":"数学","ts_ms":1652259235843}

结论：

  • 只输出两表数据流的关联数据。

3.1.2 处理时间窗口外联结

WindowInnerJoinByProcessTimeTest.java 内容如下：

  1. package cn.mfox.etl.v2.join.window.outer;
  2.  
  3. import cn.mfox.etl.v2.join.watermark.CourseDataStreamNoWatermark;
  4. import cn.mfox.etl.v2.join.watermark.TeacherDataStreamNoWatermark;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flink.api.common.functions.CoGroupFunction;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11. import org.apache.flink.util.Collector;
  12.  
  13. /**
  14.  * 根据 process time(处理时间) 进行 window outer join(窗口外关联)
  15.  * 把教师表和课程表进行窗口外联联结，关联不到的数据也会下发
  16.  * 窗口同组联结（Window CoGroup Join）
  17.  *
  18.  * @author hy
  19.  * @version 1.0
  20.  * @date 2022/5/6 16:31
  21.  */
  22. public class WindowOuterJoinByProcessTimeTest {
  23.  
  24.     public static void main(String[] args) throws Exception {
  25.         // 1.创建执行环境
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(1);
  28.         // 2.获取教师数据流和课程表数据流
  29.         DataStream<JSONObject> teacherDataStream = TeacherDataStreamNoWatermark.getTeacherDataStream(env);
  30.         DataStream<JSONObject> courseDataStream = CourseDataStreamNoWatermark.getCourseDataStream(env);
  31.         // 3.窗口联结（教师流和课程表）打印输出
  32.         windowOuterJoinAndPrint(teacherDataStream, courseDataStream);
  33.         // 4.执行任务
  34.         env.execute("WindowOuterJoinByProcessTimeTest Job");
  35.     }
  36.  
  37.     /**
  38.      * 窗口外联并打印输出
  39.      * Window上的 outer join，使用coGroup算子，关联不到的数据也会下发
  40.      *
  41.      * @param teacherDataStream 教师数据流
  42.      * @param courseDataStream  课程数据流
  43.      */
  44.     private static void windowOuterJoinAndPrint(DataStream<JSONObject> teacherDataStream,
  45.                                                 DataStream<JSONObject> courseDataStream) {
  46.         DataStream<JSONObject> teacherCourseDataStream = teacherDataStream
  47.                 .coGroup(courseDataStream)
  48.                 .where(teacher -> teacher.getString("t_id"))
  49.                 .equalTo(course -> course.getString("c_tid"))
  50.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(3L)))
  51.                 .apply(
  52.                         new CoGroupFunction<JSONObject, JSONObject, JSONObject>() {
  53.                             @Override
  54.                             public void coGroup(Iterable<JSONObject> iterable,
  55.                                                 Iterable<JSONObject> iterable1,
  56.                                                 Collector<JSONObject> collector) {
  57.                                 JSONObject result = new JSONObject();
  58.                                 for (JSONObject jsonObject : iterable) {
  59.                                     result.putAll(jsonObject);
  60.                                 }
  61.                                 for (JSONObject jsonObject : iterable1) {
  62.                                     result.putAll(jsonObject);
  63.                                 }
  64.                                 collector.collect(result);
  65.                             }
  66.                         }
  67.                 );
  68.         teacherCourseDataStream.print("Window Outer Join By Process Time");
  69.     }
  70. }

程序运行截图：

ccfda20862fd4ed5caea11b243bb6ad2.jpeg

控制台输出内容：

  1. Window Outer Join By Process Time> {"t_id":"002","op":"r","c_tid":"002","t_name":"李四","c_id":"2","c_name":"数学","ts_ms":1652259799140}
  2. Window Outer Join By Process Time> {"t_id":"001","op":"r","c_tid":"001","t_name":"张三","c_id":"1","c_name":"语文","ts_ms":1652259799140}
  3. Window Outer Join By Process Time> {"t_id":"003","op":"r","t_name":"王五","ts_ms":1652259799139}

结论：

  • 输出关联到的数据。

  • 未关联的教师表数据流也被输出（教师名称为王五的数据也被输出）。

3.1.3 处理时间窗口内外联结对比

“处理时间窗口内联结” 和 “处理时间窗口外联结” 对比：

1a65bb493a7bd732cb6e9541f4975662.jpeg
3.2 事件时间窗口
3.2.1 事件时间窗口内联结

WindowInnerJoinByEventTimeTest.java 内容如下：

  1. package cn.mfox.etl.v2.join.window.eventtime;
  2.  
  3. import cn.mfox.etl.v2.join.watermark.CourseDataStreamWithWatermark;
  4. import cn.mfox.etl.v2.join.watermark.TeacherDataStreamWithWatermark;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flink.api.common.functions.JoinFunction;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11.  
  12. /**
  13.  * 基于 事件时间 的 window inner join 把教师表和课程表进行联结
  14.  * <p>
  15.  * 只有两者数据流关联到数据，才会进行打印
  16.  *
  17.  * @author hy
  18.  * @version 1.0
  19.  * @date 2022/5/6 16:31
  20.  */
  21. public class WindowInnerJoinByEventTimeTest {
  22.  
  23.     public static void main(String[] args) throws Exception {
  24.         // 1.创建执行环境
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         // 2.获取教师数据流和课程表数据流
  28.         DataStream<JSONObject> teacherDataStream = TeacherDataStreamWithWatermark.getTeacherDataStream(env);
  29.         DataStream<JSONObject> courseDataStream = CourseDataStreamWithWatermark.getCourseDataStream(env);
  30.         // 3.窗口联结（教师流和课程表）打印输出
  31.         windowInnerJoinAndPrint(teacherDataStream, courseDataStream);
  32.         // 4.执行任务
  33.         env.execute("WindowInnerJoinByEventTimeTest Job");
  34.     }
  35.  
  36.     /**
  37.      * 窗口联结并打印输出
  38.      * 只支持 inner join，即窗口内联关联到的才会下发，关联不到的则直接丢掉。
  39.      * 如果想实现Window上的 outer join，需要使用coGroup算子
  40.      *
  41.      * @param teacherDataStream 教师数据流
  42.      * @param courseDataStream  课程数据流
  43.      */
  44.     private static void windowInnerJoinAndPrint(DataStream<JSONObject> teacherDataStream,
  45.                                                 DataStream<JSONObject> courseDataStream) {
  46.         DataStream<JSONObject> teacherCourseDataStream = teacherDataStream
  47.                 .join(courseDataStream)
  48.                 .where(teacher -> teacher.getString("t_id"))
  49.                 .equalTo(couse -> couse.getString("c_tid"))
  50.                 .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
  51.                 .apply(
  52.                         new JoinFunction<JSONObject, JSONObject, JSONObject>() {
  53.                             @Override
  54.                             public JSONObject join(JSONObject jsonObject,
  55.                                                    JSONObject jsonObject2) {
  56.                                 // 拼接
  57.                                 jsonObject.putAll(jsonObject2);
  58.                                 return jsonObject;
  59.                             }
  60.                         }
  61.                 );
  62.         teacherCourseDataStream.print("Window Inner Join By Event Time");
  63.     }
  64. }
3.2.2 事件时间窗口外联结

WindowOuterJoinByEventTimeTest.java 内容如下：

  1. package cn.mfox.etl.v2.join.window.eventtime;
  2.  
  3. import cn.mfox.etl.v2.join.watermark.CourseDataStreamWithWatermark;
  4. import cn.mfox.etl.v2.join.watermark.TeacherDataStreamWithWatermark;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flink.api.common.functions.CoGroupFunction;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11. import org.apache.flink.util.Collector;
  12.  
  13. /**
  14.  * 根据 event time(事件时间) 进行 window outer join(窗口外关联)
  15.  * 把教师表和课程表进行窗口外联联结，关联不到的数据也会下发
  16.  *
  17.  * @author hy
  18.  * @version 1.0
  19.  * @date 2022/5/6 16:31
  20.  */
  21. public class WindowOuterJoinByEventTimeTest {
  22.  
  23.     public static void main(String[] args) throws Exception {
  24.         // 1.创建执行环境
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         // 2.获取教师数据流和课程表数据流
  28.         DataStream<JSONObject> teacherDataStream = TeacherDataStreamWithWatermark.getTeacherDataStream(env);
  29.         DataStream<JSONObject> courseDataStream = CourseDataStreamWithWatermark.getCourseDataStream(env);
  30.         // 3.窗口联结（教师流和课程表）打印输出
  31.         windowOuterJoinAndPrint(teacherDataStream, courseDataStream);
  32.         // 4.执行任务
  33.         env.execute("WindowOuterJoinByEventTimeTest Job");
  34.     }
  35.  
  36.     /**
  37.      * 窗口外联并打印输出
  38.      * Window上的 outer join，使用coGroup算子，关联不到的数据也会下发
  39.      *
  40.      * @param teacherDataStream 教师数据流
  41.      * @param courseDataStream  课程数据流
  42.      */
  43.     private static void windowOuterJoinAndPrint(DataStream<JSONObject> teacherDataStream,
  44.                                                 DataStream<JSONObject> courseDataStream) {
  45.         DataStream<JSONObject> teacherCourseDataStream = teacherDataStream
  46.                 .coGroup(courseDataStream)
  47.                 .where(teacher -> teacher.getString("t_id"))
  48.                 .equalTo(course -> course.getString("c_tid"))
  49.                 .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
  50.                 .apply(
  51.                         new CoGroupFunction<JSONObject, JSONObject, JSONObject>() {
  52.                             @Override
  53.                             public void coGroup(Iterable<JSONObject> iterable,
  54.                                                 Iterable<JSONObject> iterable1,
  55.                                                 Collector<JSONObject> collector) {
  56.                                 JSONObject result = new JSONObject();
  57.                                 for (JSONObject jsonObject : iterable) {
  58.                                     result.putAll(jsonObject);
  59.                                 }
  60.                                 for (JSONObject jsonObject : iterable1) {
  61.                                     result.putAll(jsonObject);
  62.                                 }
  63.                                 collector.collect(result);
  64.                             }
  65.                         }
  66.                 );
  67.         teacherCourseDataStream.print("Window Outer Join By Event Time");
  68.     }
  69. }
3.2.3 验证案例

因 “事件时间” 是跟随数据本身的时间，所以验证的前置条件：

  • 事件时间内联结 和 事件时间外联结 程序都需要处于运行状态。

  • 教师表和课程表需要插入数据，促进水位线移动，完成事件时间窗口计算。

具体验证步骤如下：

  1. 启动 “事件时间 内联结” 程序

e970661a976d90e7774d98f9d3ae932b.jpeg

  1. 启动 “事件时间 外联结” 程序

d86a0d3e58708b407adb569a23098fbf.jpeg

  1. 两种表中插入数据，促进水位线移动（重要）

SQL 语句如下：

  1. # 插入教师数据
  2. insert into teacher value('004','马六');
  3.  
  4. # 插入课程数据
  5. insert into course value('3','英语','003');

插入SQL截图：

aa919088ab22b4490fab356ac02fe572.jpeg

  1. “事件时间 内联结” 控制台输出

输出截图：

4226f257b64fe26c9d5740aa389525a7.jpeg

控制台打印内容：

  1. Window Inner Join By Event Time> {"t_id":"001","op":"r","c_tid":"001","t_name":"张三","c_id":"1","c_name":"语文","ts_ms":1652334256893}
  2. Window Inner Join By Event Time> {"t_id":"002","op":"r","c_tid":"002","t_name":"李四","c_id":"2","c_name":"数学","ts_ms":1652334256894}

  1. “事件时间 外联结” 控制台输出

输出截图：

935d14dedb4fa359a8a95f5a54891166.jpeg

控制台打印内容：

  1. Window Outer Join By Event Time> {"t_id":"002","op":"r","c_tid":"002","t_name":"李四","c_id":"2","c_name":"数学","ts_ms":1652334264544}
  2. Window Outer Join By Event Time> {"t_id":"001","op":"r","c_tid":"001","t_name":"张三","c_id":"1","c_name":"语文","ts_ms":1652334264543}
  3. Window Outer Join By Event Time> {"t_id":"003","op":"r","t_name":"王五","ts_ms":1652334264544}

  1. “事件时间窗口内联结” 和 “事件时间窗口外联结” 对比：

109077cd8febad2f2345e45e9b83025c.jpeg

4 间隔联结

4.1 概念说明

官网说明链接：

https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/datastream/operators/joining/#interval-join

官网说明截图：

a99d75f3d86380742b4e99804b001184.jpeg
4.2 间隔联结案例

InteralJoinByEventTimeTest.java 内容如下：

  1. package cn.mfox.etl.v2.join.interal;
  2.  
  3. import cn.mfox.etl.v2.join.watermark.CourseDataStreamWithWatermark;
  4. import cn.mfox.etl.v2.join.watermark.TeacherDataStreamWithWatermark;
  5. import com.alibaba.fastjson.JSONObject;
  6. import org.apache.flink.streaming.api.datastream.DataStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  9. import org.apache.flink.streaming.api.windowing.time.Time;
  10. import org.apache.flink.util.Collector;
  11.  
  12. /**
  13.  * interal join(间隔联结) 把教师表和课程表进行联结
  14.  * 间隔联结只支持 事件时间，不支持 处理时间
  15.  *
  16.  * @author hy
  17.  * @version 1.0
  18.  * @date 2022/5/6 16:31
  19.  */
  20. public class InteralJoinByEventTimeTest {
  21.  
  22.     public static void main(String[] args) throws Exception {
  23.         // 1.创建执行环境
  24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  25.         env.setParallelism(1);
  26.         // 2.获取教师数据流和课程表数据流
  27.         DataStream<JSONObject> teacherDataStream = TeacherDataStreamWithWatermark.getTeacherDataStream(env);
  28.         DataStream<JSONObject> courseDataStream = CourseDataStreamWithWatermark.getCourseDataStream(env);
  29.         // 3.间隔联结（教师流和课程表）打印输出
  30.         intervalJoinAndPrint(teacherDataStream, courseDataStream);
  31.         // 4.执行任务
  32.         env.execute("TeacherJoinCourseTest Job");
  33.     }
  34.  
  35.     /**
  36.      * 间隔联结并打印输出
  37.      *
  38.      * @param teacherDataStream 教师数据流
  39.      * @param courseDataStream  课程数据流
  40.      */
  41.     private static void intervalJoinAndPrint(DataStream<JSONObject> teacherDataStream,
  42.                                              DataStream<JSONObject> courseDataStream) {
  43.         DataStream<JSONObject> teacherCourseDataStream = teacherDataStream
  44.                 .keyBy(teacher -> teacher.getString("t_id"))
  45.                 .intervalJoin(
  46.                         courseDataStream.keyBy(course -> course.getString("c_tid"))
  47.                 )
  48.                 .between(
  49.                         Time.seconds(-5),
  50.                         Time.seconds(5)
  51.                 )
  52.                 .process(
  53.                         new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
  54.                             @Override
  55.                             public void processElement(JSONObject left, JSONObject right,
  56.                                                        Context ctx, Collector<JSONObject> out) {
  57.                                 left.putAll(right);
  58.                                 out.collect(left);
  59.                             }
  60.                         }
  61.                 );
  62.         teacherCourseDataStream.print("Interval Join By Event Time");
  63.     }
  64. }

程序运行截图：

73feda3db6d35e8f4afcce55b74986de.jpeg

控制台输出内容：

  1. Interval Join By Event Time> {"t_id":"003","op":"r","c_tid":"003","t_name":"王五","c_id":"3","c_name":"英语","ts_ms":1652335319793}
  2. Interval Join By Event Time> {"t_id":"002","op":"r","c_tid":"002","t_name":"李四","c_id":"2","c_name":"数学","ts_ms":1652335319793}
  3. Interval Join By Event Time> {"t_id":"001","op":"r","c_tid":"001","t_name":"张三","c_id":"1","c_name":"语文","ts_ms":1652335319793}

结论：

  • 只输出关联到的数据。

  • 只支持 “事件时间” 。

  • 控制台启动后，无需插入数据促进水位线流动也能输出数据。

5 总结

针对 “双流 JOIN” 的多种方式联结，总结对比如下：

257b9a3e6076c8e40268600a2ea5edd5.jpeg

如果这个文章对你有帮助，不要忘记 「在看」 「点赞」 「收藏」 三连啊喂！

17a293e3e1bdf45947bcd76dd426c1dc.png

6bf7b65357289b799ce7b2d0e27ebf50.jpeg

2022年全网首发|大数据专家级技能模型与学习指南(胜天半子篇)

互联网最坏的时代可能真的来了

我在B站读大学，大数据专业

我们在学习Flink的时候，到底在学习什么？

193篇文章暴揍Flink，这个合集你需要关注一下

Flink生产环境TOP难题与优化，阿里巴巴藏经阁YYDS

Flink CDC我吃定了耶稣也留不住他！| Flink CDC线上问题小盘点

我们在学习Spark的时候，到底在学习什么？

在所有Spark模块中，我愿称SparkSQL为最强！

硬刚Hive | 4万字基础调优面试小总结

数据治理方法论和实践小百科全书

标签体系下的用户画像建设小指南

4万字长文 | ClickHouse基础&实践&调优全视角解析

【面试&个人成长】2021年过半，社招和校招的经验之谈

大数据方向另一个十年开启 |《硬刚系列》第一版完结

我写过的关于成长/面试/职场进阶的文章

当我们在学习Hive的时候在学习什么？「硬刚Hive续集」

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/782823
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号