您的位置: 消费周刊网 > 大数据 > 正文

Flink教程:DataStream上的Join操作

2020-03-28 06:26:33来源:阅读:

批处理经常要解决的问题是将两个数据源做关联Join操作。比如,很多手机APP都有一个用户数据源User,同时APP会记录用户的行为,我们称之为Behavior,两个表按照userId来进行Join。在流处理场景下,Flink也支持了Join,只不过Flink是在一个时间窗口上来进行两个表的Join。

Join示例图

目前,Flink支持了两种Join:Window Join(窗口连接)和Interval Join(时间间隔连接。

Window Join

从名字中能猜到,Window Join主要在Flink的窗口上进行操作,它将两个流中落在相同窗口的元素按照某个Key进行Join。一个Window Join的大致骨架结构为:

input1.join(input2)    .where(<KeySelector>)      <- input1使用哪个字段作为Key    .equalTo(<KeySelector>)    <- input2使用哪个字段作为Key    .window(<WindowAssigner>)  <- 指定WindowAssigner    [.trigger(<Trigger>)]      <- 指定Trigger(可选)    [.evictor(<Evictor>)]      <- 指定Evictor(可选)    .apply(<JoinFunction>)     <- 指定JoinFunction

下图展示了Join的大致过程。两个输入数据流先分别按Key进行分组,然后将元素划分到窗口中。窗口的划分需要使用WindowAssigner来定义,这里可以使用Flink提供的滚动窗口、滑动窗口或会话窗口等默认的WindowAssigner。随后两个数据流中的元素会被分配到各个窗口上,也就是说一个窗口会包含来自两个数据流的元素。相同窗口内的数据会以INNER JOIN的语义来相互关联,形成一个数据对。当窗口的时间结束,Flink会调用JoinFunction来对窗口内的数据对进行处理。当然,我们也可以使用Trigger或Evictor做一些自定义优化,他们的使用方法和普通窗口的使用方法一样。

Join的大致流程

接下来我们重点分析一下两个数据流是如何INNER JOIN的:

窗口内数据INNER JOIN示意图

一般滴,INNER JOIN只对两个数据源都出现的元素做Join,形成一个数据对,即数据源input1中的某个元素与数据源input2中的所有元素逐个配对。当数据源某个窗口内没数据时,比如图中的第三个窗口,Join的结果也是空的。

class MyJoinFunction extends JoinFunction[(String, Int), (String, Int), String] {  override def join(input1: (String, Int), input2: (String, Int)): String = {    "input 1 :" + input1._2 + ", input 2 :" + input2._2  }}val input1: DataStream[(String, Int)] = ...val input2: DataStream[(String, Int)] = ...val joinResult = input1.join(input2)      .where(i1 => i1._1)      .equalTo(i2 => i2._1)      .window(TumblingProcessingTimeWindows.of(Time.seconds(60)))      .apply(new MyJoinFunction)

上面的代码自定义了JoinFunction,并将Join结果打印出来。无论代码中演示的滚动窗口,还是滑动窗口或会话窗口,其原理都是一样的。除了JoinFunction,Flink还提供了FlatJoinFunction,其功能是输出零到多个结果。

如果INNER JOIN不能满足我们的需求,CoGroupFunction提供了更多可自定义的功能。需要注意的是,在调用时,要写成input1.coGroup(input2).where(<KeySelector>).equalTo(<KeySelecotr>)。

class MyCoGroupFunction extends CoGroupFunction[(String, Int), (String, Int), String] {  // 这里的类型是Java的Iterable,需要引用 collection.JavaConverters._ 并转成Scala  override def coGroup(input1: lang.Iterable[(String, Int)], input2: lang.Iterable[(String, Int)], out: Collector[String]): Unit = {    input1.asScala.foreach(element => out.collect("input1 :" + element.toString()))    input2.asScala.foreach(element => out.collect("input2 :" + element.toString()))  }}val input1: DataStream[(String, Int)] = ...val input2: DataStream[(String, Int)] = ...val coGroupResult = input1.coGroup(input2)      .where(i1 => i1._1)      .equalTo(i2 => i2._1)      .window(TumblingProcessingTimeWindows.of(Time.seconds(60)))      .apply(new MyCoGroupFunction)

Interval Join

与Window Join不同,Interval Join不依赖Flink的WindowAssigner,而是根据一个时间间隔(Interval)界定时间。Interval需要一个时间下界(lower bound)和上界(upper bound),如果我们将input1和input2进行Interval Join,input1中的某个元素为input1.element1,时间戳为input1.element1.ts,那么一个Interval就是[input1.element1.ts + lower bound, input1.element1.ts + upper bound],input2中落在这个时间段内的元素将会和input1.element1组成一个数据对。用数学公式表达为,凡是符合下面公式input1.element1.ts + lower bound <= input2.elementx.ts <=input1.element1.ts + upper bound的元素使用INNER JOIN语义,两两组合在一起。上下界可以是正数也可以是负数。

注意,目前Flink(1.9)的Interval Join只支持Event Time语义。

Interval Join示意图

下面的代码展示了如何对两个数据流进行Interval Join:

class MyProcessFunction extends ProcessJoinFunction[(String, Long, Int), (String, Long, Int), String] {  override def processElement(input1: (String, Long, Int),                              input2: (String, Long, Int),                              context: ProcessJoinFunction[(String, Long, Int), (String, Long, Int), String]#Context,                              out: Collector[String]): Unit = {    out.collect("input 1: " + input1.toString() + ", input 2: " + input2.toString)  }}// 数据流有三个字段:(key, 时间戳, 数值)val input1: DataStream[(String, Long, Int)] = ...val input2: DataStream[(String, Long, Int)] = ...val intervalJoinResult = input1.keyBy(_._1)      .intervalJoin(input2.keyBy(_._1))      .between(Time.milliseconds(-5), Time.milliseconds(10))      .process(new MyProcessFunction)

默认的时间间隔是包含上下界的,我们可以使用.lowerBoundExclusive() 和.upperBoundExclusive来确定是否需要包含上下界。

val intervalJoinResult = input1.keyBy(_._1)      .intervalJoin(input2.keyBy(_._1))      .between(Time.milliseconds(-5), Time.milliseconds(10))      .upperBoundExclusive()      .lowerBoundExclusive()      .process(new MyProcessFunction)

Interval Join内部是用缓存来存储所有数据的,因此需要注意缓存数据不能太大,以免对内存造成绝大压力。

推荐阅读:懒人洗袜子小窍门

滚动推荐
26:33Flink教程:DataStrea
.where <- input1使用哪个字段作为Key [详细]
06:53路侧停车电子收费如何缴纳?
1、如何缴纳道路停车费?(1)线上支付您可通过“北京交通”APP、微信[详细]
15:43功夫汽车丨合资车企“转型大业”如何
越是寒冬,越要苦练内功。说起来容易,做起来却殊为不易。在新冠疫情的影响[详细]
05:02专属酷炫设计,全新Jeep⁺指南者
你未必出类拔萃,但肯定与众不同。全新Jeep+指南者加料上市,所有你想[详细]
31:235G对农业意味着什么
【导语】2019年,我国正式进入5G商用元年,各种5G落地应用场[详细]
30:05错把肠癌便血当痔疮,这些惨痛教训要
林先生今年46岁,是一位“痔疮”患者,去年底还做了痔疮手术。但术[详细]
56:58聚势待发,谋远共赢|亚太天能携手创
随着社会经济水平的发展,5G时代的到来,智能家居趋向大众化普及。[详细]