经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
flink 两个datastream实现left_join的两种方法
来源：cnblogs　　作者：feiquan　　时间：2020/11/9 16:08:56　　对本文有异议

本文只是以left_join作为举例，right_join,full_join 等是同理的,大家可以自行扩展

 

1. 实验思路

  1.提供两个流

            nameStream: 用户名称信息从 9999 端口获取

            ageStream: 有用户年龄信息 从9998 端口获取

  2. 用nameStream left join  ageStream ，合并两个流中的信息，输出结果

 

2. 方法

     1【简单】： 使用 coGroup 来进行left join. 

     2【复杂】：使用 nameStream join (nameStream.union(ageStream)) 后, 按id 分组,判断分组后记录的个数。

                    偶数时，代表同一个id,会有俩次匹配，一次是left*left, 一次是 left * right, 返回非自身匹配就可以,也就是 left*right；

                    奇数时，代表只有一次 left*left 直接返回

 

3.代码

公用对象：

  1. import java.text.SimpleDateFormat
  2. import java.util.Date
  4. import com.alibaba.fastjson.JSON
  5. import com.alibaba.fastjson.serializer.SerializerFeature
  7. import scala.beans.BeanProperty
  9. object Const {
  10.   final val names = List("wang", "li", "zhao", "qian", "sun")
  11.   final val ages = List(0, 10, 20, 30)
  12.   final val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
  13. }
  15. case class People(
  16.                    @BeanProperty var id: Int = 0,
  17.                    @BeanProperty var name: String = "",
  18.                    @BeanProperty var age: Int = 0,
  19.                    @BeanProperty var eventTime: Long = 0,
  20.                    @BeanProperty var eventTimeStr: String = ""
  21.                  ) {
  23.   override def toString: String = {
  24.     JSON.toJSONString(this, SerializerFeature.QuoteFieldNames)
  25.   }
  26. }
  28. object People {
  29.   def newPeopleByName(id: Int, name: String) = {
  30.     val than = new People()
  31.     val now = new Date()
  32.     than.id = id
  33.     than.name = name
  34.     than.eventTime = now.getTime
  35.     than.eventTimeStr = Const.dateFormat.format(now)
  36.     than
  37.   }
  39.   def newPeopleByAge(id: Int, age: Int) = {
  40.     val than = new People()
  41.     val now = new Date()
  42.     than.id = id
  43.     than.age = age
  44.     than.eventTime = now.getTime
  45.     than.eventTimeStr = Const.dateFormat.format(now)
  46.     than
  47.   }
  48. }
People

 

stream 测试：

  1. import java.lang
  3. import org.apache.flink.api.common.functions.{CoGroupFunction, JoinFunction}
  4. import org.apache.flink.api.scala._
  5. import org.apache.flink.streaming.api.TimeCharacteristic
  6. import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
  7. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  8. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  9. import org.apache.flink.streaming.api.scala.function.RichWindowFunction
  10. import org.apache.flink.streaming.api.watermark.Watermark
  11. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
  12. import org.apache.flink.streaming.api.windowing.time.Time
  13. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  14. import org.apache.flink.util.Collector
  17. object DataStreamCoGroup {
  18.   def main(args: Array[String]): Unit = {
  19.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  20.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  21.     env.setParallelism(1)
  22.     val WIN = 10 // s
  24.     // id,name
  25.     val nameStream = env
  26.       //      .addSource(new NameSource)
  27.       .socketTextStream("localhost", 9999)
  28.       .filter(_.nonEmpty)
  29.       .map(x => {
  30.         val arr = x.split(",")
  31.         People.newPeopleByName(id = arr(0).toInt, name = arr(1))
  32.       })
  33.       .assignTimestampsAndWatermarks(
  34.         new BoundedOutOfOrdernessTimestampExtractor[People](Time.seconds(10L)) {
  35.           override def extractTimestamp(element: People): Long = element.eventTime
  36.         })
  38.     // id,age
  39.     val ageStream = env
  40.       //      .addSource(new AgeSource)
  41.       .socketTextStream("localhost", 9998)
  42.       .filter(_.nonEmpty)
  43.       .map(x => {
  44.         val arr = x.split(",")
  45.         People.newPeopleByAge(id = arr(0).toInt, age = arr(1).toInt)
  46.       })
  47.       .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[People] {
  48.         val maxOutOfOrderness = 1000L // 最大无序数据到达的时间，用来生成水印,10s
  49.         var currentMaxTimestamp: Long = _
  51.         override def getCurrentWatermark: Watermark = {
  52.           new Watermark(currentMaxTimestamp - maxOutOfOrderness)
  53.         }
  55.         override def extractTimestamp(element: People, previousElementTimestamp: Long): Long = {
  56.           currentMaxTimestamp = Math.max(element.eventTime, currentMaxTimestamp)
  57.           element.eventTime
  58.         }
  59.       })
  61.     nameStream
  62.       .map(("name:", _))
  63.       .print()
  65.     ageStream
  66.       .map(("age:", _))
  67.       .print()
  69.     // coGroup
  70.     nameStream
  71.       .coGroup(ageStream)
  72.       .where(_.id).equalTo(_.id)
  73.       .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
  74.       .apply(new CoGroupFunction[People, People, People] {
  75.         override def coGroup(first: lang.Iterable[People], second: lang.Iterable[People], out: Collector[People]): Unit = {
  76.           if (first.iterator().hasNext) {
  77.             first.forEach(left => {
  78.               var noElement = true
  79.               if (second.iterator().hasNext) {
  80.                 second.forEach(rigth => {
  81.                   noElement = false
  82.                   val data = left.copy()
  83.                   data.age = rigth.age
  84.                   out.collect(data)
  85.                 })
  86.               }
  87.               if (noElement) out.collect(left)
  88.             })
  89.           }
  91.         }
  92.       })
  93.       .map(("-------- coGroup left Join", _))
  94.       .print()
  96.     // -------------------------------------------
  97.     //   left join
  98.     val nameStream2 = nameStream.map((_, "nameStream"))
  99.     val ageStream2 = ageStream.map((_, "ageStream"))
  100.     //    nameStream2
  101.     //      .print()
  102.     //
  103.     //    ageStream2
  104.     //      .print()
  106.     nameStream2
  107.       .join(nameStream2.union(ageStream2))
  108.       .where(_._1.id)
  109.       .equalTo(_._1.id)
  110.       .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
  111.       .apply(new JoinFunction[Tuple2[People, String], Tuple2[People, String], Tuple2[People, Boolean]] {
  112.         override def join(first: (People, String), second: (People, String)): (People, Boolean) = {
  113.           val data = first._1.copy()
  114.           var same = true // 是否和自身连接
  115.           if (second._2.equals("ageStream")) {
  116.             same = false
  117.             data.age = second._1.age
  118.           }
  119.           (data, same)
  120.         }
  121.       })
  122.       .keyBy(_._1.id)
  123.       .window(TumblingEventTimeWindows.of(Time.seconds(WIN)))
  124.       .apply(new RichWindowFunction[Tuple2[People, Boolean], People, Int, TimeWindow] {
  125.         override def apply(key: Int, window: TimeWindow, input: Iterable[(People, Boolean)], out: Collector[People]): Unit = {
  126.           var len = 0
  127.           var same: People = null
  128.           var unSame: People = null
  129.           for (i <- input.iterator) {
  130.             len += 1
  131.             if (i._2) same = i._1.copy() // a1,a1
  132.             else unSame = i._1.copy() //  a1,a2
  133.           }
  135.           if (len % 2 == 0) out.collect(unSame) // 偶数返回 a1,a2
  136.           else out.collect(same) // 奇数返回 a1,a1
  137.         }
  138.       })
  139.       .map(("-------- left Join ", _))
  140.       .print()
  143.     env.execute("DataStreamCoGroup")
  144.   }
  145. }

 

4. 测试

   1. name 9999： [0s,10s)内输入 

  1. (name:,{"age":0,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  2. (name:,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})

 

   2.age 9998: [0s,10s)内输入(1,1),(3,3)

  1. (age:,{"age":1,"eventTime":1602951661454,"eventTimeStr":"2020-10-18 00:21:01","id":1,"name":""})
  2. (age:,{"age":2,"eventTime":1602951662457,"eventTimeStr":"2020-10-18 00:21:02","id":2,"name":""})

 

   3. 10s 过去后，再等1s 的水印延迟时间, 分别在两个流中输入任意信息，主要是为了更新水印信息触发window 计算

  1. (name:,{"age":0,"eventTime":1602951682681,"eventTimeStr":"2020-10-18 00:21:22","id":4,"name":"4"})
  2. (age:,{"age":4,"eventTime":1602951688581,"eventTimeStr":"2020-10-18 00:21:28","id":4,"name":""})

   4. 结果：

  1. (-------- coGroup left Join,{"age":1,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  2. (-------- coGroup left Join,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})
  3. (-------- left Join ,{"age":1,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  4. (-------- left Join ,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})

    可以看到，两种方式的结果一致

 

完整的输出：

  1. (age:,{"age":1,"eventTime":1602951661454,"eventTimeStr":"2020-10-18 00:21:01","id":1,"name":""})
  2. (age:,{"age":2,"eventTime":1602951662457,"eventTimeStr":"2020-10-18 00:21:02","id":2,"name":""})
  3. (name:,{"age":0,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  4. (name:,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})
  5. (name:,{"age":0,"eventTime":1602951682681,"eventTimeStr":"2020-10-18 00:21:22","id":4,"name":"4"})
  6. (age:,{"age":4,"eventTime":1602951688581,"eventTimeStr":"2020-10-18 00:21:28","id":4,"name":""})
  7. (-------- coGroup left Join,{"age":1,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  8. (-------- coGroup left Join,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})
  9. (-------- left Join ,{"age":1,"eventTime":1602951665626,"eventTimeStr":"2020-10-18 00:21:05","id":1,"name":"1"})
  10. (-------- left Join ,{"age":0,"eventTime":1602951668775,"eventTimeStr":"2020-10-18 00:21:08","id":3,"name":"3"})

 

原文链接：http://www.cnblogs.com/feiquan/p/13833256.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号