flink学习（14）—— 双流join

写过一篇  金牌会员 | 2024-12-3 04:10:18 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 769|帖子 769|积分 2307

概述

Join:内连接
CoGroup：内连接，左连接，右连接
Interval Join：点对面
Join

  1. 1、Join 将有相同 Key 并且位于同一窗口中的两条流的元素进行关联。
  2. 2、Join 可以支持处理时间（processing time）和事件时间(event time)两种时间特征。
  3. 3、Join 通用用法如下：
  4.     stream.join(otherStream)
  5.         .where(<KeySelector>)
  6.         .equalTo(<KeySelector>)
  7.         .window(<WindowAssigner>)
  8.         .apply(<JoinFunction>)
复制代码
滚动窗口

  1. package com.bigdata.day07;
  4. import org.apache.commons.lang3.time.DateUtils;
  5. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  7. import org.apache.flink.api.common.functions.JoinFunction;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.java.functions.KeySelector;
  10. import org.apache.flink.api.java.tuple.Tuple3;
  11. import org.apache.flink.streaming.api.datastream.DataStream;
  12. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  16. import org.apache.flink.streaming.api.windowing.time.Time;
  19. import java.text.ParseException;
  20. import java.time.Duration;
  21. import java.util.Date;
  23. /**
  24. * 内连接
  25. * 可以通过两个socket流，将数据合并为一个三元组，key,value1,value2
  26. */
  27. public class _01_双流join_join_内连接 {
  28.     public static void main(String[] args) throws Exception {
  29.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  30.         env.setParallelism(1);
  31.         
  32.         // 绿色的流
  33.         DataStreamSource<String> source = env.socketTextStream("localhost", 7777);
  34.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> greenSource = source.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  35.             @Override
  36.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  37.                 String[] split = line.split(",");
  38.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  39.             }
  40.         }).assignTimestampsAndWatermarks(
  41.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  42.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  43.                             @Override
  44.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  45.                                 
  46.                                 String timeStr = tuple3.f2;
  48.                                 try {
  49.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  50.                                     return date.getTime();
  51.                                 } catch (ParseException e) {
  52.                                     throw new RuntimeException(e);
  53.                                 }
  55.                             }
  56.                         }));
  57.         
  58.         // 红色的流
  59.         DataStreamSource<String> source2 = env.socketTextStream("localhost", 7778);
  60.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> redSource = source2.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  61.             @Override
  62.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  63.                 String[] split = line.split(",");
  64.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  65.             }
  66.         }).assignTimestampsAndWatermarks(
  67.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  68.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  69.                             @Override
  70.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  71.                               
  72.                                 String timeStr = tuple3.f2;
  74.                                 try {
  75.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  76.                                     return date.getTime();
  77.                                 } catch (ParseException e) {
  78.                                     throw new RuntimeException(e);
  79.                                 }
  81.                             }
  82.                         }));
  83.                         
  84.         // 双流join
  85.         DataStream<Tuple3<String, Integer, Integer>> rsSource = greenSource.join(redSource).where(new KeySelector<Tuple3<String, Integer, String>, String>() {
  86.             @Override
  87.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  88.                 return tuple3.f0;
  89.             }
  90.         }).equalTo(new KeySelector<Tuple3<String, Integer, String>, String>() {
  91.             @Override
  92.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  93.                 return tuple3.f0;
  94.             }
  95.             // 滚动窗口
  96.         }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
  97.                 .apply(new JoinFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, Tuple3<String, Integer, Integer>>() {
  98.             @Override
  99.             public Tuple3<String, Integer, Integer> join(Tuple3<String, Integer, String> first, Tuple3<String, Integer, String> second) throws Exception {
  100.                 return Tuple3.of(first.f0, first.f1, second.f1);
  101.             }
  102.         });
  104.         redSource.print("红色的流：");
  105.         greenSource.print("绿色的流：");
  106.         rsSource.print("合并后的流：");
  108.         env.execute();
  111.     }
  112. }
复制代码
滑动窗口

  1. package com.bigdata.day07;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.JoinFunction;
  7. import org.apache.flink.api.common.functions.MapFunction;
  8. import org.apache.flink.api.java.tuple.Tuple3;
  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.datastream.SingleOutputStreamOperator;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
  14. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  15. import org.apache.flink.streaming.api.windowing.time.Time;
  17. import java.text.ParseException;
  18. import java.text.SimpleDateFormat;
  19. import java.time.Duration;
  20. import java.util.Date;
  23. /**
  24. * @基本功能: 演示join的滑动窗口
  25. * @program:FlinkDemo
  26. * @author: 闫哥
  27. * @create:2024-05-20 09:11:13
  28. **/
  29. public class Demo02Join {
  31.     public static void main(String[] args) throws Exception {
  33.         //1. env-准备环境
  34.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  35.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
  36.         // 将并行度设置为1,否则很难看到现象
  37.         env.setParallelism(1);
  38.         // 创建一个绿色的流
  39.         DataStreamSource<String> greenSource = env.socketTextStream("localhost", 8899);
  40.         // key,0,2021-03-26 12:09:00 将它变为三元组
  41.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> greenDataStream = greenSource.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  42.             @Override
  43.             public Tuple3<String, Integer, String> map(String value) throws Exception {
  44.                 String[] arr = value.split(",");
  45.                 return new Tuple3<>(arr[0], Integer.valueOf(arr[1]), arr[2]);
  46.             }
  47.         }).assignTimestampsAndWatermarks(
  48.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  49.                         // 为什么这个地方的代码比之前要长，原因是以前获取的数据都是long类型，并且都是毫秒值
  50.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  51.                             @Override
  52.                             public long extractTimestamp(Tuple3<String, Integer, String> element, long recordTimestamp) {
  53.                                 // 指定你的数据中哪一个是时间戳，并且时间戳必须是long类型，必须是毫秒为单位的。
  54.                                 String time = element.f2; //2021-03-26 12:09:00
  55.                                 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  56.                                 try {
  57.                                     Date date = sdf.parse(time);
  58.                                     return date.getTime();
  59.                                 } catch (ParseException e) {
  60.                                     throw new RuntimeException(e);
  61.                                 }
  62.                             }
  63.                         })
  64.         );
  65.         // 创建一个橘色的流
  66.         DataStreamSource<String> orangeSource = env.socketTextStream("localhost", 9988);
  67.         // key,0,2021-03-26 12:09:00 将它变为三元组
  68.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> orangeDataStream = orangeSource.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  69.             @Override
  70.             public Tuple3<String, Integer, String> map(String value) throws Exception {
  71.                 String[] arr = value.split(",");
  72.                 return new Tuple3<>(arr[0], Integer.valueOf(arr[1]), arr[2]);
  73.             }
  74.         }).assignTimestampsAndWatermarks(
  75.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  76.                         // 为什么这个地方的代码比之前要长，原因是以前获取的数据都是long类型，并且都是毫秒值
  77.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  78.                             @Override
  79.                             public long extractTimestamp(Tuple3<String, Integer, String> element, long recordTimestamp) {
  80.                                 // 指定你的数据中哪一个是时间戳，并且时间戳必须是long类型，必须是毫秒为单位的。
  81.                                 String time = element.f2; //2021-03-26 12:09:00
  82.                                 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  83.                                 try {
  84.                                     Date date = sdf.parse(time);
  85.                                     return date.getTime();
  86.                                 } catch (ParseException e) {
  87.                                     throw new RuntimeException(e);
  88.                                 }
  89.                             }
  90.                         })
  91.         );
  93.         //2. source-加载数据
  94.         //3. transformation-数据处理转换
  95.         DataStream<Tuple3<String, Integer, Integer>> resultStream = greenDataStream.join(orangeDataStream)
  96.                 .where(tuple3 -> tuple3.f0)
  97.                 .equalTo(tuple3 -> tuple3.f0)
  98.                 // 滑动窗口
  99.                 .window(SlidingEventTimeWindows.of(Time.seconds(5),Time.seconds(1)))
  100.                 .apply(new JoinFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, Tuple3<String, Integer, Integer>>() {
  101.                     @Override
  102.                     public Tuple3<String, Integer, Integer> join(Tuple3<String, Integer, String> first, Tuple3<String, Integer, String> second) throws Exception {
  103.                         return Tuple3.of(first.f0, first.f1, second.f1);
  104.                     }
  105.                 });
  106.         //4. sink-数据输出
  107.         greenDataStream.print("绿色的流：");
  108.         orangeDataStream.print("橘色的流：");
  110.         resultStream.print("最终的结果：");
  113.         //5. execute-执行
  114.         env.execute();
  115.     }
  116. }
复制代码
CoGroup

  1. 1、优势：可以实现内连接，左连接，右连接
  2. 2、劣势：内存压力大
  3. 3、和上面的写法区别：将join换成coGroup，apply中实现的具体方法有区别
  4. 4、流程
  5. stream.coGroup(otherStream)
  6.         .where(<KeySelector>)
  7.         .equalTo(<KeySelector>)
  8.         .window(<WindowAssigner>)
  9.         .apply(<CoGroupFunction>);
复制代码
内连接

  1. package com.bigdata.day07;
  4. import org.apache.commons.lang3.time.DateUtils;
  5. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  7. import org.apache.flink.api.common.functions.CoGroupFunction;
  8. import org.apache.flink.api.common.functions.JoinFunction;
  9. import org.apache.flink.api.common.functions.MapFunction;
  10. import org.apache.flink.api.java.functions.KeySelector;
  11. import org.apache.flink.api.java.tuple.Tuple3;
  12. import org.apache.flink.streaming.api.datastream.DataStream;
  13. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  14. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  17. import org.apache.flink.streaming.api.windowing.time.Time;
  18. import org.apache.flink.util.Collector;
  20. import java.text.ParseException;
  21. import java.time.Duration;
  22. import java.util.Date;
  24. /**
  25. * 内连接
  26. */
  27. public class _02_双流join_CoGroup_内连接 {
  28.     public static void main(String[] args) throws Exception {
  29.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  30.         env.setParallelism(1);
  32.         // 绿色的流
  33.         DataStreamSource<String> source = env.socketTextStream("localhost", 7777);
  34.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> greenSource = source.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  35.             @Override
  36.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  37.                 String[] split = line.split(",");
  38.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  39.             }
  40.         }).assignTimestampsAndWatermarks(
  41.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  42.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  43.                             @Override
  44.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  45.                               
  46.                                 String timeStr = tuple3.f2;
  48.                                 try {
  49.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  50.                                     return date.getTime();
  51.                                 } catch (ParseException e) {
  52.                                     throw new RuntimeException(e);
  53.                                 }
  55.                             }
  56.                         }));
  57.                         
  58.         // 红色的流
  59.         DataStreamSource<String> source2 = env.socketTextStream("localhost", 7778);
  60.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> redSource = source2.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  61.             @Override
  62.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  63.                 String[] split = line.split(",");
  64.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  65.             }
  66.         }).assignTimestampsAndWatermarks(
  67.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  68.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  69.                             @Override
  70.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  71.                                 String timeStr = tuple3.f2;
  73.                                 try {
  74.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  75.                                     return date.getTime();
  76.                                 } catch (ParseException e) {
  77.                                     throw new RuntimeException(e);
  78.                                 }
  80.                             }
  81.                         }));
  82.         
  83.         
  84.         // 连接
  85.         DataStream<Tuple3<String, String, String>> rsSource = greenSource.coGroup(redSource).where(new KeySelector<Tuple3<String, Integer, String>, String>() {
  86.             @Override
  87.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  88.                 return tuple3.f0;
  89.             }
  90.         }).equalTo(new KeySelector<Tuple3<String, Integer, String>, String>() {
  91.             @Override
  92.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  93.                 return tuple3.f0;
  94.             }
  95.         }).window(TumblingEventTimeWindows.of(Time.seconds(5))).apply(new CoGroupFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, Tuple3<String, String, String>>() {
  96.             @Override
  97.             public void coGroup(Iterable<Tuple3<String, Integer, String>> first, Iterable<Tuple3<String, Integer, String>> second, Collector<Tuple3<String, String, String>> out) throws Exception {
  98.                 for (Tuple3<String, Integer, String> firesTuple3 : first) {
  99.                     for (Tuple3<String, Integer, String> secondTuple3 : second) {
  100.                         out.collect(Tuple3.of(firesTuple3.f0,"green"+firesTuple3.f1,"red"+secondTuple3.f1));
  101.                     }
  102.                 }
  103.             }
  104.         });
  106.         redSource.print("红色的流：");
  107.         greenSource.print("绿色的流：");
  108.         rsSource.print("合并后的流：");
  110.         env.execute();
  113.     }
  114. }
复制代码
外连接

  1. package com.bigdata.day07;
  4. import org.apache.commons.lang3.time.DateUtils;
  5. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  7. import org.apache.flink.api.common.functions.CoGroupFunction;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.java.functions.KeySelector;
  10. import org.apache.flink.api.java.tuple.Tuple3;
  11. import org.apache.flink.streaming.api.datastream.DataStream;
  12. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  16. import org.apache.flink.streaming.api.windowing.time.Time;
  17. import org.apache.flink.util.Collector;
  19. import java.text.ParseException;
  20. import java.time.Duration;
  21. import java.util.Date;
  23. /**
  24. * 外连接
  25. */
  26. public class _03_双流join_CoGroup_外连接 {
  27.     public static void main(String[] args) throws Exception {
  28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  29.         env.setParallelism(1);
  30.         
  31.         // 绿色的流
  32.         DataStreamSource<String> source = env.socketTextStream("localhost", 7777);
  33.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> greenSource = source.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  34.             @Override
  35.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  36.                 String[] split = line.split(",");
  37.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  38.             }
  39.         }).assignTimestampsAndWatermarks(
  40.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  41.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  42.                             @Override
  43.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  44.                                 String timeStr = tuple3.f2;
  46.                                 try {
  47.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  48.                                     return date.getTime();
  49.                                 } catch (ParseException e) {
  50.                                     throw new RuntimeException(e);
  51.                                 }
  53.                             }
  54.                         }));
  55.         // 红色的流
  56.         DataStreamSource<String> source2 = env.socketTextStream("localhost", 7778);
  57.         SingleOutputStreamOperator<Tuple3<String, Integer, String>> redSource = source2.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  58.             @Override
  59.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  60.                 String[] split = line.split(",");
  61.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  62.             }
  63.         }).assignTimestampsAndWatermarks(
  64.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  65.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  66.                             @Override
  67.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  68.                                 String timeStr = tuple3.f2;
  70.                                 try {
  71.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  72.                                     return date.getTime();
  73.                                 } catch (ParseException e) {
  74.                                     throw new RuntimeException(e);
  75.                                 }
  77.                             }
  78.                         }));
  79.         DataStream<Tuple3<String, String, String>> rsSource = greenSource.coGroup(redSource).where(new KeySelector<Tuple3<String, Integer, String>, String>() {
  80.             @Override
  81.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  82.                 return tuple3.f0;
  83.             }
  84.         }).equalTo(new KeySelector<Tuple3<String, Integer, String>, String>() {
  85.             @Override
  86.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  87.                 return tuple3.f0;
  88.             }
  89.         }).window(TumblingEventTimeWindows.of(Time.seconds(5))).apply(new CoGroupFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, Tuple3<String, String, String>>() {
  90.             @Override
  91.             public void coGroup(Iterable<Tuple3<String, Integer, String>> first, Iterable<Tuple3<String, Integer, String>> second, Collector<Tuple3<String, String, String>> out) throws Exception {
  92.             // 内连接，左连接，右连接的区别只在这里面存在，两层循环
  93.                 for (Tuple3<String, Integer, String> firesTuple3 : first) {
  94.                     boolean isExist = false;
  95.                     for (Tuple3<String, Integer, String> secondTuple3 : second) {
  96.                         isExist = true;
  97.                         out.collect(Tuple3.of(firesTuple3.f0,"green"+firesTuple3.f1,"red"+secondTuple3.f1));
  98.                     }
  99.                     if (!isExist){
  100.                         out.collect(Tuple3.of(firesTuple3.f0,"green"+firesTuple3.f1,"red null"));
  101.                     }
  102.                 }
  103.             }
  104.         });
  106.         redSource.print("红色的流：");
  107.         greenSource.print("绿色的流：");
  108.         rsSource.print("合并后的流：");
  110.         env.execute();
  113.     }
  114. }
复制代码
Interval Join


  1. 1、Join以及CoGroup 原因是 Join和CoGroup是窗口Join，必须给定窗口
  2. 2、Interval Join不需要给窗口。Interval Join 必须先分组才能使用。
  3. 3、先对数据源进行keyBy
  4. 4、 外流.intervalJoin（内流）
  5.         .between(-2,2)
  6.         .process
  8. between 左不包，右包
  9. 内部的流为下面的流（取单个值）
复制代码
 代码实现

  1. package com.bigdata.day07;
  4. import org.apache.commons.lang3.time.DateUtils;
  5. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  6. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  7. import org.apache.flink.api.common.functions.CoGroupFunction;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.java.functions.KeySelector;
  10. import org.apache.flink.api.java.tuple.Tuple3;
  11. import org.apache.flink.streaming.api.datastream.DataStream;
  12. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  13. import org.apache.flink.streaming.api.datastream.KeyedStream;
  14. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  17. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  18. import org.apache.flink.streaming.api.windowing.time.Time;
  19. import org.apache.flink.util.Collector;
  21. import java.text.ParseException;
  22. import java.time.Duration;
  23. import java.util.Date;
  26. public class _04_双流join_Interval_Join {
  27.     public static void main(String[] args) throws Exception {
  28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  29.         env.setParallelism(1);
  30.         
  31.         //绿色的流
  32.         DataStreamSource<String> source = env.socketTextStream("localhost", 7777);
  33.         
  34.         KeyedStream<Tuple3<String, Integer, String>, String> greenSource = source.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  35.             @Override
  36.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  37.                 String[] split = line.split(",");
  38.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  39.             }
  40.             // 水印
  41.         }).assignTimestampsAndWatermarks(
  42.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  43.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  44.                             @Override
  45.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  46.                                 String timeStr = tuple3.f2;
  48.                                 try {
  49.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  50.                                     return date.getTime();
  51.                                 } catch (ParseException e) {
  52.                                     throw new RuntimeException(e);
  53.                                 }
  55.                             }
  56.                             // keyBy
  57.                         })).keyBy(new KeySelector<Tuple3<String, Integer, String>, String>() {
  58.             @Override
  59.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  60.                 return tuple3.f0;
  61.             }
  62.         });
  65.         // 红色的流
  66.         DataStreamSource<String> source2 = env.socketTextStream("localhost", 7778);
  67.         KeyedStream<Tuple3<String, Integer, String>, String> redSource = source2.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
  68.             @Override
  69.             public Tuple3<String, Integer, String> map(String line) throws Exception {
  70.                 String[] split = line.split(",");
  71.                 return Tuple3.of(split[0], Integer.valueOf(split[1]), split[2]);
  72.             }
  73.             // 水印
  74.         }).assignTimestampsAndWatermarks(
  75.                 WatermarkStrategy.<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  76.                         .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
  77.                             @Override
  78.                             public long extractTimestamp(Tuple3<String, Integer, String> tuple3, long recordTimestamp) {
  79.                                 String timeStr = tuple3.f2;
  81.                                 try {
  82.                                     Date date = DateUtils.parseDate(timeStr, "yyyy-MM-dd hh-mm-ss");
  83.                                     return date.getTime();
  84.                                 } catch (ParseException e) {
  85.                                     throw new RuntimeException(e);
  86.                                 }
  88.                             }
  89.                             // 分组
  90.                         })).keyBy(new KeySelector<Tuple3<String, Integer, String>, String>() {
  91.             @Override
  92.             public String getKey(Tuple3<String, Integer, String> tuple3) throws Exception {
  93.                 return tuple3.f0;
  94.             }
  95.         });
  96.         
  97.         // 实现
  98.         SingleOutputStreamOperator<String> rsSource = greenSource
  99.                 .intervalJoin(redSource)
  100.                 .between(Time.seconds(-2), Time.seconds(2))
  101.                 .process(new ProcessJoinFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, String>() {
  102.             @Override
  103.             public void processElement(Tuple3<String, Integer, String> left, Tuple3<String, Integer, String> right, ProcessJoinFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, String>.Context ctx, Collector<String> out) throws Exception {
  104.                 out.collect("left中的key:"+left.f0+",value="+left.f1+",time="+left.f2+",right中的key:"+right.f0+",value="+right.f1+",time="+right.f2);
  105.             }
  106.         });
  110.         redSource.print("红色的流：");
  111.         greenSource.print("绿色的流：");
  112.         rsSource.print("合并后的流：");
  114.         env.execute();
  115. /**
  116. * 红色的为下面的流
  117. * 范围：
  118. * 假如现在是10
  119. * 9 10 11 12
  120. */
  122.     }
  123. }
复制代码


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

写过一篇

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表