正文

Flink流处理系列（六）flink实现多数据源水位线的inner join

Rae
Rae V管理员 /0 评论 /559 阅读
0313

在前面的文章我们主要介绍了flink多数据源的时间窗口，然后对数据进行计算，但是试想一下，真实的生产环境中我们经常多数据源都是乱序的，所以此时我们多个数据源拿过来进行join的时候，由于数据处理的延迟，会导致如下情况：

同时有A和B两批数据发进来，时间窗口设置为1分组，按照推算在10:10-10:11分之间A和B分别都会有10条数据进来，但是由于乱序的原因，本来B的提前发了其他数据，
导致本来B应该在10:10分到10:11分发送10条数据的，但是真实的情况确实B在10:10分到10:11分只发了8条数据，剩下两条数据在10:11分到10:12分才发过来。

此时按照常规的窗口计算的话，本来应该处理10条数据，结果只处理了8条数据，剩下两条数据在后面的时间过来的，结果前面窗口的数据已经被销毁了，导致数据就join不起来，最终的结果也就不准确了。所以这时候怎么办呢？这就是这里提到的水位线。

在flink中，水位线的意思是：

在事件时间的语义下，不依赖系统时间，而是基于数据自带的时间戳去定义一个时钟，用来表示当前时间的进展。

在数据流中加入一个时钟标记，记录当前的事件时间，这个标记可以直接广播到下游，当下游任务收到这个标记，就可以更新自己的时钟了，这种类似于水流中用来做标志的记号，在Flink中被称为水位线

水位线的效果就是：

假设场景，比如说上学校车，校车每天早上九点发车，但有一部分学生可能九点零二才能赶来，于是小明偷偷把司机的时间调后了两秒，这样当时间到了九点（延后两秒的九点）大家都能上车了。

所以按照我的理解就是原本时间窗口是10点10分到10点11分，这边利用水位线的话，就相当于给时间种调慢一点，原本到10点11分进行一次结算，如果设置了水位线为5秒，则真实的情况是在10点11分5秒的时候进行一次结算，这样子就尽量整合完所有的数据。

下面我们还是按照惯例来演示下，还是使用前面的案例

一、定义学生的水位线接收stream

package com.flink.demo.connector;

import java.time.Duration;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.alibaba.fastjson.JSONObject;
import com.flink.demo.utils.TransformUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class UserDataWithWaterStream {

	public static DataStream<JSONObject> getUserDataStream(StreamExecutionEnvironment env) {
		// 1.创建Flink-MySQL-CDC的Source
		MySqlSource<String> teacherSouce = MySqlSource.<String>builder().hostname("192.168.31.20").port(33307)
				.username("root").password("123456").databaseList("users").tableList("users.user")
				.startupOptions(StartupOptions.initial()).deserializer(new JsonDebeziumDeserializationSchema())
				.serverTimeZone("Asia/Shanghai").build();

		// 2.使用CDC Source从MySQL读取数据
		DataStreamSource<String> mysqlDataStreamSource = env.fromSource(teacherSouce,
				WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
						.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
							/**
							 * 
							 */
							private static final long serialVersionUID = 1L;

							@Override
							public long extractTimestamp(String extractData, long l) {
								return JSONObject.parseObject(extractData).getLong("ts_ms");
							}
						}),
				"UserDataStreamNoWatermark Source");
		// 3.转换为指定格式
		DataStream<JSONObject> teacherDataStream = mysqlDataStreamSource.map(rawData -> {
			return TransformUtil.formatResult(rawData);
		});
		return teacherDataStream;
	}

}

二、定义学校的水位线接收stream

package com.flink.demo.connector;

import java.time.Duration;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.alibaba.fastjson.JSONObject;
import com.flink.demo.utils.TransformUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class SchoolDataWithWaterStream {

	public static DataStream<JSONObject> getSchoolDataStream(StreamExecutionEnvironment env) {
		// 1.创建Flink-MySQL-CDC的Source
		MySqlSource<String> teacherSouce = MySqlSource.<String>builder().hostname("192.168.31.20").port(33307)
				.username("root").password("123456").databaseList("users").tableList("users.school")
				.startupOptions(StartupOptions.initial()).deserializer(new JsonDebeziumDeserializationSchema())
				.serverTimeZone("Asia/Shanghai").build();

		// 2.使用CDC Source从MySQL读取数据
		DataStreamSource<String> mysqlDataStreamSource = env.fromSource(teacherSouce,
				WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
						.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
							/**
							 * 
							 */
							private static final long serialVersionUID = 1L;

							@Override
							public long extractTimestamp(String extractData, long l) {
								return JSONObject.parseObject(extractData).getLong("ts_ms");
							}
						}),
				"SchoolDataStreamNoWatermark Source");
		// 3.转换为指定格式
		DataStream<JSONObject> teacherDataStream = mysqlDataStreamSource.map(rawData -> {
			return TransformUtil.formatResult(rawData);
		});
		return teacherDataStream;
	}

}

三、最后编写job

package com.flink.demo.mysql;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.flink.demo.connector.SchoolDataStream;
import com.flink.demo.connector.SchoolDataWithWaterStream;
import com.flink.demo.connector.UserDataStream;
import com.flink.demo.connector.UserDataWithWaterStream;

public class MySqlCdcReaderWithWater {

	public static void main(String[] args) throws Exception {
		// 1.创建执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		// 2.1获取学生数据流
		DataStream<JSONObject> userDataStream = UserDataWithWaterStream.getUserDataStream(env);
		// 2.2获取学校数据流
		DataStream<JSONObject> schoolDataStream = SchoolDataWithWaterStream.getSchoolDataStream(env);
		// 3.窗口联结（学生流和学校流）打印输出
		windowInnerJoinAndPrint(userDataStream, schoolDataStream);
		// 4.执行任务
		env.execute("Time Window Out Join Job");
	}

	/**
	 * 只支持 inner join，即窗口内联关联到的才会下发，关联不到的则直接丢掉。
	 *
	 * @param userDataStream   用户数据流
	 * @param schoolDataStream 学校数据流
	 */
	private static void windowInnerJoinAndPrint(DataStream<JSONObject> userDataStream,
			DataStream<JSONObject> schoolDataStream) {

		DataStream<JSONObject> studentsDataStream = userDataStream.join(schoolDataStream)
				.where(user -> user.getString("school_id")).equalTo(school -> school.getString("id"))
				.window(TumblingEventTimeWindows.of(Time.seconds(10L)))
				.apply(new JoinFunction<JSONObject, JSONObject, JSONObject>() {
					/**
					 * 
					 */
					private static final long serialVersionUID = -8027510260773592640L;

					@Override
					public JSONObject join(JSONObject jsonObject, JSONObject jsonObject2) {
						// 拼接
						jsonObject.putAll(jsonObject2);
						return jsonObject;
					}
				});

		studentsDataStream.print("Window Out Join user and school");
	}
}

四、测试

这里我们准备两条sql，掐着时间点执行

INSERT INTO `users`.`user` (`id`, `name`, `age`, `sex`, `school_id`) VALUES (41, '王八', 2, '2', 42);
INSERT INTO `users`.`school` (`id`, `school_name`, `school_address`) VALUES (42, '北京十中', '石景山');

然后我们能看到迟到的数据也能被join上

image.png

以上就是我们介绍的利用水位线来操作数据流。


备注：

1、使用水位线的话需要自己定义一个时间戳。

2、需要在接收stream的时候，设置水位线的延迟，本案例的核心代码如下：

WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
						.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
							/**
							 * 
							 */
							private static final long serialVersionUID = 1L;

							@Override
							public long extractTimestamp(String extractData, long l) {
								return JSONObject.parseObject(extractData).getLong("ts_ms");
							}
						})


最后按照惯例附上本案例的源码，登录后即可下载。

登录访问
本站用户 免费查看
登录账号
您未登录，请登录 或 注册后查看
我从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。
-- 展开阅读全文 --