Flink基本概念和算子利用

基础概念

Flink是一个框架和分布式处置惩罚引擎，用于对无界数据流和有界数据流举行有状态计算，它的核心目标是“数据流上的有状态计算”。

有界流和无界流

• 有界流：具有明白的开始和结束时间，数据量有限。适合利用批处置惩罚技术，可以在处置惩罚前将全部数据一次性读入内存举行处置惩罚。有界流通常用于历史数据分析、数据迁徙等场景。‌
  
• 无界流：没有明白的开始和结束时间，数据接二连三天生。由于数据是无穷且持续的，无界流需要实时处置惩罚，并且必须持续摄取和处置惩罚数据，不能等待全部数据到达后再举行处置惩罚。适合适用于流处置惩罚。
  

名词

源算子（source）

Flink可以从各种来源获取数据，然后构建DataStream举行转换处置惩罚。一般将数据的输入来源称为数据源（data source），而读取数据的算子就是源算子（source operator）。

2. // 创建执行环境
3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
5. // 从集合读取数据
6. DataStreamSource<Integer> collectionSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15));
8. // 从文件中读取数据
9. DataStreamSource<String> fileSource = env.fromSource(FileSource.forRecordStreamFormat(new TextLineInputFormat(),new Path("input/word.txt")).build(),WatermarkStrategy.noWatermarks(),"fileSource");
12. //  从kafka读取数据
13. DataStreamSource<String> kafkaSource = env.fromSource(KafkaSource.<String>builder()
14.         // kafka地址--可配置多个
15.         .setBootstrapServers("")
16.         // topic名称--可配置多个
17.         .setTopics("")
18.         // 消费组id
19.         .setGroupId("")
20.         // 反序列化方式
21.         .setValueOnlyDeserializer(new SimpleStringSchema())
22.         // kafka 消费偏移量 方式 earliest（一定从最早的开始消费）、latest（一定从最新的开始消费）或者手动设置偏移量 ，默认是earliest
23.         .setStartingOffsets(OffsetsInitializer.latest())
24.         // 水位线，自定义数据源算子名称
25.         .build(), WatermarkStrategy.noWatermarks(), "kafkaSource");
27. // 从socket读取数据
28. DataStreamSource<String> socketSource = env.socketTextStream("...","1234");

复制代码

基本转换算子

Map

对元素的数据类型和内容做转换。

1. // 第一个参数为输入流，第二个参数为输出流
2. SingleOutputStreamOperator<UserDto> userDataStream = kafkaSource.map(new MapFunction<String, UserDto>() {   
3.     @Override     
4.     public UserDto map(String message) throws Exception {        
5.         return JSONObject.parseObject(message, UserDto.class);   
6.     }
7. });

复制代码

FlatMap

输入一个元素同时产生零个、一个或多个元素。

1. // 第一个参数为输入流，第二个参数为输出流
2. // 可做转换，可做条件过滤
3. SingleOutputStreamOperator<UserDto> userDataStream =kafkaSource.flatMap(new FlatMapFunction<String, UserDto>() {   
4.     @Override   
5.     public void flatMap(String message, Collector<UserDto> collector) throws Exception {        
6.         UserDto userDto = JSONObject.parseObject(message, UserDto.class);        
7.         collector.collect(userDto);
8.     }
9. });

复制代码

Filter

对数据源根据条件过滤数据，保留满足条件的数据

2. // 过滤出年龄大于18的用户
3. SingleOutputStreamOperator<UserDto> filterDataStream = userDataStream.filter(new FilterFunction<UserDto>() {   
4.      @Override   
5.      public boolean filter(UserDto userDto) throws Exception {        
6.         return userDto.getAge() > 18;   
7.      }
8. });

复制代码

聚合算子

KeyBy

根据指定的字段(key)，将数据划分到不相交的分区中。相同key的元素会被分到同一个分区中。

2. // 将用户id一样的用户分到一个分区内
3. KeyedStream<UserDto, Integer> userKeyedStream = userDataStream.keyBy(new KeySelector<UserDto, Integer>() {   
4.     @Override   
5.     public Integer getKey(UserDto userDto) throws Exception {        
6.         return userDto.getId();   
7.     }
8. });

复制代码

Reduce （仅支持同类型的数据）

对流的数据，来一条计算一条，将当前元素和上一次聚合后的数据组合，输出新值，并将新值举行保存，作为下一次计算的元素。
聚合前和聚合后的数据类型是一致的。
当第一条数据进来时，不会触发计算。

2. // 计算一个用户的订单总价格
3. SingleOutputStreamOperator<UserDto> reduce = userKeyedStream.reduce(new ReduceFunction<UserDto>() {
4.     @Override
5.     public UserDto reduce(UserDto t1, UserDto t2) throws Exception {
6.         int totalPrice = t1.getTotalPrice() + t2.getOrderPrice();
7.         UserDto userDto = new UserDto();
8.         userDto.setId(t1.getId());
9.         userDto.setAge(t1.getAge());
10.         userDto.setTotalPrice(totalPrice);
11.         return userDto;
12.     }
13. });

复制代码

Aggregate （支持差异类型的数据）

2. SingleOutputStreamOperator<String> aggregate = windowedStream.aggregate(new AggregateFunction<UserDto, Integer, String>() {   
3.     /**     
4.     * 创建累加器，就是初始化累加器     
5.     * @return     
6.     */   
7.     @Override   
8.     public Integer createAccumulator() {      
9.         return 0;   
10.     }   
11.     /**     
12.     * 计算逻辑或者是聚合逻辑     
13.     * @param userDto     
14.     * @param beforeData     
15.     * @return     
16.     */   
17.     @Override   
18.     public Integer add(UserDto userDto, Integer beforeData) {        
19.         return beforeData + userDto.getAge();   
20.     }   
21.     /**     
22.     * 获取最终结果，窗口触发时输出     
23.     * @param integer     
24.     * @return     
25.     */   
26.     @Override   
27.     public String getResult(Integer integer) {        
28.         return "计算结束，最终结果为：" + integer.toString();   
29.     }   
30.     /**     
31.     * 只有会话窗口才会使用到     
32.     * @param integer     
33.     * @param acc1     
34.     * @return     
35.     */   
36.     @Override   
37.     public Integer merge(Integer integer, Integer acc1) {        
38.         return 0;   
39.     }
40. });

复制代码

窗口（window）

把流切割成有限巨细的多个“存储桶”；每个数据都会分发到对应的桶中，当到达窗口结束时间时，就对每个桶中收集的数据举行计算处置惩罚。窗口不是静态天生的，是动态创建的。当这个窗口范围的进入第一条数据时，才会创建对应的窗口。
滚动窗口

有固定的巨细，是一种对数据举行“匀称切片”的划分方式。窗口之间没有重叠，也不会有间隔，是“首尾相接”的状态。每个数据都会分配到一个窗口，而且只会属于一个窗口。滚动窗口可以基于时间定义，也可以基于数据的个数定义，需要的参数只有一个，就是窗口的巨细。

1. // 分组
2. KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);
4. // 基于处理时间开窗，窗口长度为10s，窗口开始时间为 窗口长度整数倍向下取整，结束时间为开始时间+窗口长度
5. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> tumblingProcessingTimeStream = keyByStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
7. // 基于事件时间开窗，窗口长度为10s，窗口开始时间为数据源事件时间，结束时间为开始时间+窗口长度
8. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> tumblingEventTimeStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
10. // 基于次数开窗
11. WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindowStream = keyedStream.countWindow(10);

复制代码

滑动窗口

巨细是固定的，但是窗口之间不是收尾相接的，而是可以“错开”肯定的位置。定义滑动窗口的参数有2个：窗口巨细和滑动步长，滑动步长代表了窗口计算的频率。因此，如果 slide 小于窗口巨细，滑动窗口可以答应窗口重叠。这种情况下，一个元素可能会被分发到多个窗口。

1. // 分组
2. KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);
4. // 基于处理事件开窗，窗口长度为10s，滑动步长为1s
5. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> slidingProcessingTimeWindowStream = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(1)));
7. // 基于事件事件开窗，窗口长度为10s，滑动步长为1s
8. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> slidingEventTimeWindowStream = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(1)));
10. // 基于次数开窗
11. WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindowStream = keyedStream.countWindow(10, 1);

复制代码

会话窗口

是基于会话来对数据举行分组的。会话窗口只能基于时间来定义。会话窗口中，最重要的参数就是会话的超时时间，也就是两个会话窗口之间的最小距离。如果相邻两个数据到来的时间间隔（Gap）小于指定的巨细（size），那说明还在保持会话，他们就属于同一个窗口；如果gap大于Size，那么新来的数据就应该属于新的会话窗口，而前一个窗口就应该关闭了。会话窗口的长度不固定，起始和结束时间也不是确定的，各个分区之间窗口没有任何关联。会话窗口之间肯定不会重叠的，而且会保留至少size的间隔。

1. // 分组
2. KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);
4. // 基于处理时间开窗，会话间隔时间为10s
5. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> sessionWindow = keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)));
7. // 基于事件时间开窗，会话间隔时间为10s
8. WindowedStream<Tuple2<String, Integer>, String, TimeWindow> sessionWindow = keyedStream.window(EventTimeSessionWindows.withGap(Time.seconds(10)));

复制代码

全局窗口

这种窗口全局有效，会把相同key的全部数据都分配到同一个窗口中。这种窗口没有结束的时间，默认是不会触发计算的。如果盼望它能对数据举行计算，还需要自定义“触发器”（Trigger）。全局窗口没有结束的时间点，所以一般在盼望做更加机动的窗口处置惩罚时自定义利用。Flink中的计数窗口底层就是用全局窗口实现的。
窗口触发器（trigger）

定义了窗口何时被触发并决定触发后的行为（如举行窗口数据的计算或清理）。
EventTimeTrigger

基于变乱时间和水印机制来触发窗口计算。当窗口的最大时间戳小于等于当前的水印时，立刻触发窗口计算。
ProcessingTimeTrigger

基于处置惩罚时间（即呆板的体系时间）来触发窗口计算。当处置惩罚时间达到窗口的结束时间时，触发窗口计算。
CountTrigger

根据窗口内元素的数量来触发计算。当窗口内的元素数量达到预设的阈值时，触发窗口计算。
关键方法

• onElement(T element, long timestamp, W window, TriggerContext ctx)
  当元素被添加到窗口时调用，用于注册定时器或更新窗口状态。
  
• onEventTime(long time, W window, TriggerContext ctx)
  当变乱时间计时器触发时调用，用于处置惩罚变乱时间相关的触发逻辑。
  
• onProcessingTime(long time, W window, TriggerContext ctx)
  当处置惩罚时间计时器触发时调用，用于处置惩罚处置惩罚时间相关的触发逻辑。
  
• onMerge(W window, OnMergeContext ctx)
  当两个窗口合并时调用，用于合并窗口的状态和定时器。
  
• clear(W window, TriggerContext ctx)
  当窗口被删除时调用，用于清理窗口的状态和定时器。
  

1. @Override
2. public TriggerResult onElement(BatteryRuntimeFlinkDto batteryRuntimeDto, long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {   
3.     ReducingState<Long> countState = triggerContext.getPartitionedState(countStateDescriptor);   
4.    
5. }
7. @Override
8. public TriggerResult onProcessingTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {   
9.     log.info("窗口清除定时器触发，清除计数器和定时器，并关窗");   
10.     this.clear(globalWindow, triggerContext);   
11.     return TriggerResult.PURGE;
12. }
14. @Override
15. public TriggerResult onEventTime(long time, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {   
16.     return TriggerResult.CONTINUE;
17. }
19. @Override
20. public void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {   
21.     // 清除计数器   
22.     triggerContext.getPartitionedState(countStateDescriptor).clear();   
23.     // 清除定时器   
24.     triggerContext.deleteProcessingTimeTimer(triggerContext.getPartitionedState(processTimerDescription).get());
25. }

复制代码

处置惩罚算子（process）

ProcessFunction

最基本的处置惩罚函数，基于DataStream直接调用.process()时作为参数传入。

1. public class CabinetDetailProcessFunction extends ProcessFunction<CabinetDetailDto, BatteryPutTakeLogDataSourceDto> {        
2.     //往redis中写入   
3.     private transient RedisService redisService;   
4.     private String platform;   
5.     public CabinetDetailProcessFunction(String platform) {        
6.         this.platform = platform;   
7.     }   
8.     @Override   
9.     public void open(Configuration parameters) throws Exception {        
10.         super.open(parameters);        
11.         this.redisService = ApplicationContextHolder.getBean(RedisService.class);   
12.     }   
14.         @Override   
15.         public void processElement(CabinetDetailDto cabinetDetailDto, Context context, Collector<BatteryPutTakeLogDataSourceDto> collector) throws Exception {        
16.             
17.         }
18. }

复制代码

KeyedProcessFunction

对流按键分区后的处置惩罚函数，基于KeyedStream调用.process()时作为参数传入。
ProcessWindowFunction

开窗之后的处置惩罚函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。

1. public class BatteryRuntimeProcessFunction extends ProcessWindowFunction<BatteryRuntimeFlinkDto, BatteryRuntimeFlinkDto, String, GlobalWindow> {   
3. @Override   
4. public void process(String s, Context context, Iterable<BatteryRuntimeFlinkDto> iterable, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {        
5.     List<BatteryRuntimeFlinkDto> batteryRuntimeDtos = new ArrayList<>();      
6.     iterable.forEach(p -> batteryRuntimeDtos.add(p));      
7.     if (CollectionUtils.isEmpty(batteryRuntimeDtos)) {            
8.         return;        
9.     }      
10.     BatteryRuntimeFlinkDto batteryRuntimeFlinkDto =
11.     batteryRuntimeDtos.get(0);            
12.     collector.collect(batteryRuntimeFlinkDto);   
13. }}

复制代码

ProcessAllWindowFunction

同样是开窗之后的处置惩罚函数，基于AllWindowedStream调用.process()时作为参数传入。
CoProcessFunction

合并（connect）两条流之后的处置惩罚函数，基于ConnectedStreams调用.process()时作为参数传入。
ProcessJoinFunction

间隔毗连（interval join）两条流之后的处置惩罚函数，基于IntervalJoined调用.process()时作为参数传入。
BroadcastProcessFunction

广播毗连流处置惩罚函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播毗连流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做毗连（conncet）之后的产物。

2. public class BatteryRuntimeConnectProcessFunction extends BroadcastProcessFunction<BatteryRuntimeDto, BatteryPutTakeLogDataSourceDto, BatteryRuntimeFlinkDto> {
3. // 状态
4. MapStateDescriptor<String, BatteryInBoxStatusDto> descriptor =   new MapStateDescriptor<>("boxInStatus", BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<BatteryInBoxStatusDto>(){}));   
6. @Override
7. public void processElement(BatteryRuntimeDto batteryRuntimeDto, ReadOnlyContext readOnlyContext, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {   
8. // dosometing
9. }
11. @Override   
12. public void processBroadcastElement(BatteryPutTakeLogDataSourceDto batteryPutTakeLogDataSourceDto, Context context, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {
13. // dosometing
14. }

复制代码

KeyedBroadcastProcessFunction

按键分区的广播毗连流处置惩罚函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction差异的是，这时的广播毗连流，是一个KeyedStream与广播流（BroadcastStream）做毗连之后的产物。
输出算子（sink）

输出算子，就是经过一系列处置惩罚算子后的数据输出到某个位置。例如：kafka，redis，数据库等等。
KafkaSink

2. DataStream stream...;
3. KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
4. // 指定 kafka 的地址和端口
5. .setBootstrapServers("kafka地址和端口")
6. // 指定序列化器：指定Topic名称、具体的序列化
7. .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("topic名称") .setValueSerializationSchema(new SimpleStringSchema()) .build() )
8. /**
9. * EXACTLY_ONCE: 精准一次投送。这是最严格，最理想的数据投送保证。数据不丢失不重复。
10. * AT_LEAST_ONCE: 至少一次投送。数据保证不丢失，但可能会重复。
11. * NONE: 无任何额外机制保证。数据有可能丢失或者重复。
12. */
13. // sink设置保证级别为 至少一次投送。数据保证不丢失，但可能会重复
14. .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .build(); stream.sinkTo(kafkaSink);

复制代码

JDBCSink

2. DataStream<UserDto> reduceStream...;
3. // 构建jdbc sink
4. SinkFunction<UserDto> jdbcSink = JdbcSink.sink(
5. // 数据插入sql语句
6. "insert into user (`name`, `age`) values(?, ?)",
7. new JdbcStatementBuilder<UserDto>() {
8. @Override
9. // 字段映射配置
10. public void accept(PreparedStatement pStmt, UserDto userDto) throws SQLException {
11. pStmt.setString(1, userDto.getUserName());
12. pStmt.setInt(2, userDto.getAge()); } },
13. JdbcExecutionOptions
14. .builder()
15. // 批次大小,条数
16. .withBatchSize(10)
17. // 批次最大等待时间
18. .withBatchIntervalMs(5000)
19. // 重复次数
20. .withMaxRetries(1) .build(),
21. // jdbc信息配置
22. new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
23. .withDriverName("com.mysql.jdbc.Driver")
24. .withUrl("数据库地址")
25. .withUsername("root")
26. .withPassword("password")
27. .build() );
28. // 添加jdbc sink
29. reduceStream.addSink(jdbcSink);

复制代码

其他方式的sink: File、MongoDB、RabbitMQ、Elasticsearch、Apache Pulsar 等利用方式，可参考官方文档（Apache Flink Documentation）。
Flink 相关依靠：

2. <dependency>
3.     <groupId>org.apache.flink</groupId>
4.     <artifactId>flink-java</artifactId>
5.     <version>1.17.0</version>
6. </dependency>
7. <dependency>
8.     <groupId>org.apache.flink</groupId>
9.     <artifactId>flink-streaming-java</artifactId>
10.     <version>1.17.0</version>
11. </dependency>
12. <dependency>
13.     <groupId>org.apache.flink</groupId>
14.     <artifactId>flink-runtime-web</artifactId>
15.     <version>1.17.0</version>
16. </dependency>
17. <dependency>
18.     <groupId>org.apache.flink</groupId>
19.     <artifactId>flink-clients</artifactId>
20.     <version>1.17.0</version>
21. </dependency>
22. <!-- File连接器 -->
23. <dependency>
24.     <groupId>org.apache.flink</groupId>
25.     <artifactId>flink-connector-files</artifactId>
26.     <version>1.17.0</version>
27. </dependency>
28. <!-- kafka连接器 -->
29. <dependency>
30.     <groupId>org.apache.flink</groupId>
31.     <artifactId>flink-connector-kafka</artifactId>
32.     <version>1.17.0</version>
33. </dependency>
34. <!-- jdbc连接器 -->
35. <dependency>
36.     <groupId>org.apache.flink</groupId>
37.     <artifactId>flink-connector-jdbc</artifactId>
38.     <version>1.16.0</version>
39. </dependency>

复制代码

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