Flink基本概念和算子使用

基础概念

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

有界流和无界流

• 有界流：具有明确的开始和结束时间，数据量有限。适合使用批处理技术，可以在处理前将所有数据一次性读入内存进行处理。有界流通常用于历史数据分析、数据迁移等场景。‌
• 无界流：没有明确的开始和结束时间，数据连续不断生成。由于数据是无限且持续的，无界流需要实时处理，并且必须持续摄取和处理数据，不能等待所有数据到达后再进行处理。适合适用于流处理。

名词

源算子（source）

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

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 从集合读取数据
DataStreamSource<Integer> collectionSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15));

// 从文件中读取数据
DataStreamSource<String> fileSource = env.fromSource(FileSource.forRecordStreamFormat(new TextLineInputFormat(),new Path("input/word.txt")).build(),WatermarkStrategy.noWatermarks(),"fileSource");

//  从kafka读取数据
DataStreamSource<String> kafkaSource = env.fromSource(KafkaSource.<String>builder()
        // kafka地址--可配置多个
        .setBootstrapServers("")
        // topic名称--可配置多个
        .setTopics("")
        // 消费组id
        .setGroupId("")
        // 反序列化方式
        .setValueOnlyDeserializer(new SimpleStringSchema())
        // kafka 消费偏移量 方式 earliest（一定从最早的开始消费）、latest（一定从最新的开始消费）或者手动设置偏移量 ，默认是earliest
        .setStartingOffsets(OffsetsInitializer.latest())
        // 水位线，自定义数据源算子名称
        .build(), WatermarkStrategy.noWatermarks(), "kafkaSource");

// 从socket读取数据
DataStreamSource<String> socketSource = env.socketTextStream("...","1234");

基本转换算子

Map

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

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

FlatMap

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

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

Filter

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

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

聚合算子

KeyBy

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

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

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

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

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

Aggregate （支持不同类型的数据）

SingleOutputStreamOperator<String> aggregate = windowedStream.aggregate(new AggregateFunction<UserDto, Integer, String>() {    
    /**     
    * 创建累加器，就是初始化累加器     
    * @return     
    */    
    @Override    
    public Integer createAccumulator() {       
        return 0;   
    }    
    /**     
    * 计算逻辑或者是聚合逻辑     
    * @param userDto     
    * @param beforeData     
    * @return     
    */    
    @Override    
    public Integer add(UserDto userDto, Integer beforeData) {        
        return beforeData + userDto.getAge();    
    }    
    /**     
    * 获取最终结果，窗口触发时输出     
    * @param integer     
    * @return     
    */    
    @Override    
    public String getResult(Integer integer) {        
        return "计算结束，最终结果为：" + integer.toString();    
    }    
    /**     
    * 只有会话窗口才会使用到     
    * @param integer     
    * @param acc1     
    * @return     
    */    
    @Override   
    public Integer merge(Integer integer, Integer acc1) {        
        return 0;    
    }
});

合流

将来源不同的多条流，将它们的数据进行联合处理

联合（Union）

直接将多条流合在一起，要求必须流中的数据类型必须相同，合并之后的新流会包括所有流中的元素，数据类型不变。

stream1.union(stream2, stream3, ...)

连接（Connect）

将不同两条不同类型的流合在一起，但内部仍保持各自的数据形式不变，彼此之间是相互独立的。要想得到新的DataStream，还需要进一步定义一个“同处理”（co-process）转换操作，用来说明对于不同来源、不同类型的数据，怎样分别进行处理转换、得到统一的输出类型。

DataStreamSource<Integer> stream1 ...;
DataStreamSource<String> stream1 ...;

ConnectedStreams<String, Integer> connectStream = stream2.connect(stream1);

// 参数1：第一个流的数据类型，参数2：第二个流的数据类型，参数3：输出的数据类型
SingleOutputStreamOperator<String> outputStreamOperator = connectStream.map(new CoMapFunction<String, Integer, String>() {
    @Override
    public String map1(String s) throws Exception {
        return s;
    }

    @Override
    public String map2(Integer integer) throws Exception {
        return integer.toString();
    }
});

/**
 * 实现互相匹配的效果：
 * 1、两条流，不一定谁先到
 * 2、每条流有数据来，存在一个变量中
 * 3、每条流有数据来的时候，除了存变量，去另一条流存到变量去查找是否有匹配上的
 */
SingleOutputStreamOperator<String> coProcessStream = connect.process(new CoProcessFunction<String, Integer, String>() {
    List<String> list = new ArrayList<>();
    List<Integer> list2 = new ArrayList<>();

    @Override
    public void processElement1(String s, Context context, Collector<String> collector) throws Exception {
        if (!list.contains(s)) {
            list.add(s);
        }
        if (list2.contains(s)) {
            collector.collect("s1" + s);
        }

    }

    @Override
    public void processElement2(Integer integer, Context context, Collector<String> collector) throws Exception {
        // 同上
    }
});

窗口（window）

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

滚动窗口

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

// 分组
KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);

// 基于处理时间开窗，窗口长度为10s，窗口开始时间为 窗口长度整数倍向下取整，结束时间为开始时间+窗口长度
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> tumblingProcessingTimeStream = keyByStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

// 基于事件时间开窗，窗口长度为10s，窗口开始时间为数据源事件时间，结束时间为开始时间+窗口长度
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> tumblingEventTimeStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));

// 基于次数开窗
WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindowStream = keyedStream.countWindow(10);

滑动窗口

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

// 分组
KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);

// 基于处理事件开窗，窗口长度为10s，滑动步长为1s 
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> slidingProcessingTimeWindowStream = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(1)));

// 基于事件事件开窗，窗口长度为10s，滑动步长为1s
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> slidingEventTimeWindowStream = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(1)));

// 基于次数开窗
WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindowStream = keyedStream.countWindow(10, 1);

会话窗口

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

// 分组
KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(p -> p.f0);

// 基于处理时间开窗，会话间隔时间为10s
WindowedStream<Tuple2<String, Integer>, String, TimeWindow> sessionWindow = keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)));

// 基于事件时间开窗，会话间隔时间为10s
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) 当窗口被删除时调用，用于清理窗口的状态和定时器。

@Override
public TriggerResult onElement(BatteryRuntimeFlinkDto batteryRuntimeDto, long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {    
    ReducingState<Long> countState = triggerContext.getPartitionedState(countStateDescriptor);   
    
}

@Override
public TriggerResult onProcessingTime(long l, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {    
    log.info("窗口清除定时器触发，清除计数器和定时器，并关窗");    
    this.clear(globalWindow, triggerContext);    
    return TriggerResult.PURGE;
}

@Override
public TriggerResult onEventTime(long time, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {    
    return TriggerResult.CONTINUE;
}

@Override
public void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {    
    // 清除计数器    
    triggerContext.getPartitionedState(countStateDescriptor).clear();    
    // 清除定时器    
    triggerContext.deleteProcessingTimeTimer(triggerContext.getPartitionedState(processTimerDescription).get());
}

处理算子（process）

ProcessFunction

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

public class CabinetDetailProcessFunction extends ProcessFunction<CabinetDetailDto, BatteryPutTakeLogDataSourceDto> {        
    //往redis中写入    
    private transient RedisService redisService;    
    private String platform;   
    public CabinetDetailProcessFunction(String platform) {        
        this.platform = platform;    
    }    
    @Override    
    public void open(Configuration parameters) throws Exception {        
        super.open(parameters);        
        this.redisService = ApplicationContextHolder.getBean(RedisService.class);    
    }    

    @Override    
    public void processElement(CabinetDetailDto cabinetDetailDto, Context context, Collector<BatteryPutTakeLogDataSourceDto> collector) throws Exception {        
        
    }
}

KeyedProcessFunction

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

ProcessWindowFunction

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

public class BatteryRuntimeProcessFunction extends ProcessWindowFunction<BatteryRuntimeFlinkDto, BatteryRuntimeFlinkDto, String, GlobalWindow> {    

@Override   
public void process(String s, Context context, Iterable<BatteryRuntimeFlinkDto> iterable, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {        
    List<BatteryRuntimeFlinkDto> batteryRuntimeDtos = new ArrayList<>();       
    iterable.forEach(p -> batteryRuntimeDtos.add(p));       
    if (CollectionUtils.isEmpty(batteryRuntimeDtos)) {            
        return;        
    }       
    BatteryRuntimeFlinkDto batteryRuntimeFlinkDto = 
    batteryRuntimeDtos.get(0);             
    collector.collect(batteryRuntimeFlinkDto);    
}}

ProcessAllWindowFunction

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

CoProcessFunction

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

ProcessJoinFunction

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

BroadcastProcessFunction

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

public class BatteryRuntimeConnectProcessFunction extends BroadcastProcessFunction<BatteryRuntimeDto, BatteryPutTakeLogDataSourceDto, BatteryRuntimeFlinkDto> {
// 状态
MapStateDescriptor<String, BatteryInBoxStatusDto> descriptor =   new MapStateDescriptor<>("boxInStatus", BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<BatteryInBoxStatusDto>(){}));    

@Override
public void processElement(BatteryRuntimeDto batteryRuntimeDto, ReadOnlyContext readOnlyContext, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {   
// dosometing
}

@Override    
public void processBroadcastElement(BatteryPutTakeLogDataSourceDto batteryPutTakeLogDataSourceDto, Context context, Collector<BatteryRuntimeFlinkDto> collector) throws Exception {
// dosometing
}

KeyedBroadcastProcessFunction

按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。

输出算子（sink）

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

KafkaSink

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

JDBCSink

DataStream<UserDto> reduceStream...; 
// 构建jdbc sink 
SinkFunction<UserDto> jdbcSink = JdbcSink.sink( 
// 数据插入sql语句 
"insert into user (`name`, `age`) values(?, ?)", 
new JdbcStatementBuilder<UserDto>() { 
@Override 
// 字段映射配置 
public void accept(PreparedStatement pStmt, UserDto userDto) throws SQLException { 
pStmt.setString(1, userDto.getUserName()); 
pStmt.setInt(2, userDto.getAge()); } }, 
JdbcExecutionOptions
.builder() 
// 批次大小,条数
.withBatchSize(10) 
// 批次最大等待时间 
.withBatchIntervalMs(5000) 
// 重复次数
.withMaxRetries(1) .build(),
// jdbc信息配置
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withDriverName("com.mysql.jdbc.Driver")
.withUrl("数据库地址")
.withUsername("root") 
.withPassword("password")
.build() ); 
// 添加jdbc sink 
reduceStream.addSink(jdbcSink);

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

Flink 相关依赖：

<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-java</artifactId> 
    <version>1.17.0</version> 
</dependency>
<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-streaming-java</artifactId>
    <version>1.17.0</version> 
</dependency>
<dependency> 
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-runtime-web</artifactId>
    <version>1.17.0</version>
</dependency>
<dependency> 
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>1.17.0</version> 
</dependency>
<!-- File连接器 --> 
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId> 
    <version>1.17.0</version>
</dependency>
<!-- kafka连接器 --> 
<dependency> 
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId> 
    <version>1.17.0</version> 
</dependency>
<!-- jdbc连接器 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc</artifactId> 
    <version>1.16.0</version>
</dependency>