【Flink】多流转换

多流转换

分流基本合流操作

联合（Union）连接（Connect）广播连接流 基于时间的合流——双流联结

窗口联结（Window Join）间隔联结（Interval Join）窗口同组联结（Window coGroup）

就是基于侧输出流

// 定义侧输出流标签
OutputTag outputTag = new OutputTag("side-output"){};

public void processElement(){
	// 转换成Long， 输出到主流中
	out.collect(Long.valueof(value));

	// 转换成String， 输出到侧输出流中
	ctx.output(outputTag, String.valueof(value));
}

// 获得侧输出流
DataStream stringStream = longStream.getSideOutput(outputTag);

Union：要求数据类型一样

stream1.union(stream2, stream3...)

如果流的水位线延迟时间不一样，上游两条流，下游一条流，下游会以上游水位线小的那个为准，为了不漏掉数据。

只能针对两条流。

通过connect得到连接流，然后通过map等操作得到类型一致的一条流，像一国两制。

stream1.connect(stream2)
	.map(new CoMapFunction(){
			String map1(Long value){}
			String map2(Integer vlaue){}
		})
	.print();

实时对账：
用户在APP中付款，APP产生一条流（支付日志），第三方支付平台产生一条流，按照单号连接起来。

代码待敲。

流1：数据流，主要的业务数据
流2：配置项的流，针对下游所有并行子任务都生效
这样，就可以允许配置项变动。
广播流的实现原理：将配置项放到类似Map数据结构的状态中，把这个状态广播到下游。

ruleStateDescriptor = new MapStateDescriptor<>(...);
ruleBroadcastStream = ruleStream.broadcast(ruleStateDescriptor);
dataStream.connect(ruleBroadcastStream)

数据量大的放前面，数据量小的放后面。在相同的时间窗口中可以join。不同窗口的内容不搭腔。

stream1大.join(stream2小)
	.where(data -> data.f0)
	.equalTo(data -> data.f0)
	.window(TumblingEventTimeWindows.of(Time.seconds(5)))
	.apply(new JoinFunction, Tuple2, String>(){
		@Override
		public String join(Tuple2 first, Tuple2 second) throws Exception{
			return first + "->"+second;
		}
	}).print();

下方的流A去匹配上方的流B，所以基于A的每个数据的元素，都可以开辟一个间隔区间。

流A和流B也必须基于相同的key。

下界<=上界，两者都可正可负。

a.timestamp + 下界 <= b.timestamp <= a.timestamp + 上界。

间隔联结目前只支持事件时间语义。

streamA.keyBy()
	.intervalJoin(streamB.keyBy())
	.between(下界，上界)
	.process(new ProcessJoinFunction<...>{ 
		void processElement(...){...}
	})

//统计每个用户下单前5秒和后10s的事件
orderStream.keyBy(用户名)
	.intervalJoin(clickStream.keyBy(用户名))
	.between(Time.seconds(-5), Time.seconds(10))
	.process(new ProcessJoinFunction(){
		void processElement(){
			...	
		}
	})

与window join窗口联结几乎一样：

stream1大.coGroup(stream2小)
	.where(data -> data.f0)
	.equalTo(data -> data.f0)
	.window(TumblingEventTimeWindows.of(Time.seconds(5)))
	.apply(new CoGroupFunction, Tuple2, String>(){
		@Override
		public String coGroup(Iterable> first, Iterable> second) throws Exception{
			return first + "->"+second;
		}
	}).print();

coGroup是更加通用的联结方式，可以实现内连接、外连接。

内连接：连接结果仅包含符合连接条件的行组合起来作为结果集，参与连接的两个表都应该符合连接条件。使用关键词：INNER JOIN 连接多张表。

外连接 ：连接结果不仅包含符合连接条件的行，同时也包含自身不符合条件的行。 包括左外连接、右外连接和全外连接。 左外连接 ：左边表数据行全部保留，右边表保留符合连接条件的行。