Flink处理函数实战之三:KeyedProcessFunction类

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
简介: 通过实战学习和了解处理函数的KeyedProcessFunction类

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos

Flink处理函数实战系列链接

  1. 深入了解ProcessFunction的状态操作(Flink-1.10)
  2. ProcessFunction
  3. KeyedProcessFunction类
  4. ProcessAllWindowFunction(窗口处理)
  5. CoProcessFunction(双流处理)

本篇概览

本文是《Flink处理函数实战》系列的第三篇,上一篇《Flink处理函数实战之二:ProcessFunction类》学习了最简单的ProcessFunction类,今天要了解的KeyedProcessFunction,以及该类带来的一些特性;

关于KeyedProcessFunction

通过对比类图可以确定,KeyedProcessFunction和ProcessFunction并无直接关系:
在这里插入图片描述
KeyedProcessFunction用于处理KeyedStream的数据集合,相比ProcessFunction类,KeyedProcessFunction拥有更多特性,官方文档如下图红框,状态处理和定时器功能都是KeyedProcessFunction才有的:
在这里插入图片描述
介绍完毕,接下来通过实例来学习吧;

版本信息

  1. 开发环境操作系统:MacBook Pro 13寸, macOS Catalina 10.15.3
  2. 开发工具:IDEA ULTIMATE 2018.3
  3. JDK:1.8.0_211
  4. Maven:3.6.0
  5. Flink:1.9.2

    源码下载

    如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blog_demos):
名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址,https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址,ssh协议

这个git项目中有多个文件夹,本章的应用在flinkstudy文件夹下,如下图红框所示:
在这里插入图片描述

实战简介

本次实战的目标是学习KeyedProcessFunction,内容如下:

  1. 监听本机9999端口,获取字符串;
  2. 将每个字符串用空格分隔,转成Tuple2实例,f0是分隔后的单词,f1等于1;
  3. 上述Tuple2实例用f0字段分区,得到KeyedStream;
  4. KeyedSteam转入自定义KeyedProcessFunction处理;
  5. 自定义KeyedProcessFunction的作用,是记录每个单词最新一次出现的时间,然后建一个十秒的定时器,十秒后如果发现这个单词没有再次出现,就把这个单词和它出现的总次数发送到下游算子;

    编码

  6. 继续使用《Flink处理函数实战之二:ProcessFunction类》一文中创建的工程flinkstudy;
  7. 创建bean类CountWithTimestamp,里面有三个字段,为了方便使用直接设为public:
    ```java
    package com.bolingcavalry.keyedprocessfunction;

public class CountWithTimestamp {
public String key;

public long count;

public long lastModified;

}

3. 创建FlatMapFunction的实现类Splitter,作用是将字符串分割后生成多个Tuple2实例,f0是分隔后的单词,f1等于1:
```java
package com.bolingcavalry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;

public class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
    @Override
    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {

        if(StringUtils.isNullOrWhitespaceOnly(s)) {
            System.out.println("invalid line");
            return;
        }

        for(String word : s.split(" ")) {
            collector.collect(new Tuple2<String, Integer>(word, 1));
        }
    }
}
  1. 最后是整个逻辑功能的主体:ProcessTime.java,这里面有自定义的KeyedProcessFunction子类,还有程序入口的main方法,代码在下面列出来之后,还会对关键部分做介绍:
    ```java
    package com.bolingcavalry.keyedprocessfunction;

import com.bolingcavalry.Splitter;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**

  • @author will
  • @email zq2599@gmail.com
  • @date 2020-05-17 13:43
  • @description 体验KeyedProcessFunction类(时间类型是处理时间)
    */
    public class ProcessTime {

    /**

    • KeyedProcessFunction的子类,作用是将每个单词最新出现时间记录到backend,并创建定时器,
    • 定时器触发的时候,检查这个单词距离上次出现是否已经达到10秒,如果是,就发射给下游算子
      */
      static class CountWithTimeoutFunction extends KeyedProcessFunction, Tuple2> {

      // 自定义状态
      private ValueState state;

      @Override
      public void open(Configuration parameters) throws Exception {

       // 初始化状态,name是myState
       state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
      

      }

      @Override
      public void processElement(

           Tuple2<String, Integer> value,
           Context ctx,
           Collector<Tuple2<String, Long>> out) throws Exception {
      
       // 取得当前是哪个单词
       Tuple currentKey = ctx.getCurrentKey();
      
       // 从backend取得当前单词的myState状态
       CountWithTimestamp current = state.value();
      
       // 如果myState还从未没有赋值过,就在此初始化
       if (current == null) {
           current = new CountWithTimestamp();
           current.key = value.f0;
       }
      
       // 单词数量加一
       current.count++;
      
       // 取当前元素的时间戳,作为该单词最后一次出现的时间
       current.lastModified = ctx.timestamp();
      
       // 重新保存到backend,包括该单词出现的次数,以及最后一次出现的时间
       state.update(current);
      
       // 为当前单词创建定时器,十秒后后触发
       long timer = current.lastModified + 10000;
      
       ctx.timerService().registerProcessingTimeTimer(timer);
      
       // 打印所有信息,用于核对数据正确性
       System.out.println(String.format("process, %s, %d, lastModified : %d (%s), timer : %d (%s)\n\n",
               currentKey.getField(0),
               current.count,
               current.lastModified,
               time(current.lastModified),
               timer,
               time(timer)));
      

      }

      /**

      • 定时器触发后执行的方法
      • @param timestamp 这个时间戳代表的是该定时器的触发时间
      • @param ctx
      • @param out
      • @throws Exception
        */
        @Override
        public void onTimer(

         long timestamp,
         OnTimerContext ctx,
         Collector<Tuple2<String, Long>> out) throws Exception {
        

        // 取得当前单词
        Tuple currentKey = ctx.getCurrentKey();

        // 取得该单词的myState状态
        CountWithTimestamp result = state.value();

        // 当前元素是否已经连续10秒未出现的标志
        boolean isTimeout = false;

        // timestamp是定时器触发时间,如果等于最后一次更新时间+10秒,就表示这十秒内已经收到过该单词了,
        // 这种连续十秒没有出现的元素,被发送到下游算子
        if (timestamp == result.lastModified + 10000) {

         // 发送
         out.collect(new Tuple2<String, Long>(result.key, result.count));
        
         isTimeout = true;
        

        }

        // 打印数据,用于核对是否符合预期
        System.out.println(String.format("ontimer, %s, %d, lastModified : %d (%s), stamp : %d (%s), isTimeout : %s\n\n",

             currentKey.getField(0),
             result.count,
             result.lastModified,
             time(result.lastModified),
             timestamp,
             time(timestamp),
             String.valueOf(isTimeout)));
        

        }
        }

public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 并行度1
    env.setParallelism(1);

    // 处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

    // 监听本地9999端口,读取字符串
    DataStream<String> socketDataStream = env.socketTextStream("localhost", 9999);

    // 所有输入的单词,如果超过10秒没有再次出现,都可以通过CountWithTimeoutFunction得到
    DataStream<Tuple2<String, Long>> timeOutWord = socketDataStream
            // 对收到的字符串用空格做分割,得到多个单词
            .flatMap(new Splitter())
            // 设置时间戳分配器,用当前时间作为时间戳
            .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {

                @Override
                public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
                    // 使用当前系统时间作为时间戳
                    return System.currentTimeMillis();
                }

                @Override
                public Watermark getCurrentWatermark() {
                    // 本例不需要watermark,返回null
                    return null;
                }
            })
            // 将单词作为key分区
            .keyBy(0)
            // 按单词分区后的数据,交给自定义KeyedProcessFunction处理
            .process(new CountWithTimeoutFunction());

    // 所有输入的单词,如果超过10秒没有再次出现,就在此打印出来
    timeOutWord.print();

    env.execute("ProcessFunction demo : KeyedProcessFunction");
}

public static String time(long timeStamp) {
    return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));
}

}
```
上述代码有几处需要重点关注的:

  1. 通过assignTimestampsAndWatermarks设置时间戳的时候,getCurrentWatermark返回null,因为用不上watermark;
  2. processElement方法中,state.value()可以取得当前单词的状态,state.update(current)可以设置当前单词的状态,这个功能的详情请参考《深入了解ProcessFunction的状态操作(Flink-1.10)》
  3. registerProcessingTimeTimer方法设置了定时器的触发时间,注意这里的定时器是基于processTime,和官方demo中的eventTime是不同的;
  4. 定时器触发后,onTimer方法被执行,里面有这个定时器的全部信息,尤其是入参timestamp,这是原本设置的该定时器的触发时间;

    验证

  5. 在控制台执行命令nc -l 9999,这样就可以从控制台向本机的9999端口发送字符串了;
  6. 在IDEA上直接执行ProcessTime类的main方法,程序运行就开始监听本机的9999端口了;
  7. 在前面的控制台输入aaa,然后回车,等待十秒后,IEDA的控制台输出以下信息,从结果可见符合预期:
    在这里插入图片描述
  8. 继续输入aaa再回车,连续两次,中间间隔不要超过10秒,结果如下图,可见每一个Tuple2元素都有一个定时器,但是第二次输入的aaa,其定时器在出发前,aaa的最新出现时间就被第三次输入的操作给更新了,于是第二次输入aaa的定时器中的对比操作发现此时距aaa的最近一次(即第三次)出现还未达到10秒,所以第二个元素不会发射到下游算子:
    在这里插入图片描述
  9. 下游算子收到的所有超时信息会打印出来,如下图红框,只打印了数量等于1和3的记录,等于2的时候因为在10秒内再次输入了aaa,因此没有超时接收,不会在下游打印:
    在这里插入图片描述
    至此,KeyedProcessFunction处理函数的学习就完成了,其状态读写和定时器操作都是很实用能力,希望本文可以给您提供参考;

欢迎关注阿里云开发者社区:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴...

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
在数字化转型中,企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00,阿里云三位技术专家将讲解Kafka与Flink的强强联合方案,帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景,具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师,参与还有机会领取定制好礼!扫描海报二维码或点击链接预约直播:[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
340 35
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
168 11
|
10月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
698 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
6月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
探索Flink动态CEP:杭州银行的实战案例
210 5
|
9月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写,介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库,支持在不重启服务的情况下动态更新规则,适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用,并展示了某金融机构的实际应用案例。通过动态CEP,用户可以实时调整规则,提高系统的灵活性和响应速度,降低维护成本。文中还提供了具体的代码示例和技术细节,帮助读者理解和使用Flink动态CEP。
1106 2
探索Flink动态CEP:杭州银行的实战案例
|
9月前
|
数据处理 数据安全/隐私保护 流计算
Flink 三种时间窗口、窗口处理函数使用及案例
Flink 是处理无界数据流的强大工具,提供了丰富的窗口机制。本文介绍了三种时间窗口(滚动窗口、滑动窗口和会话窗口)及其使用方法,包括时间窗口的概念、窗口处理函数的使用和实际案例。通过这些机制,可以灵活地对数据流进行分析和计算,满足不同的业务需求。
1006 27
|
11月前
|
SQL 消息中间件 分布式计算
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
大数据-115 - Flink DataStream Transformation 多个函数方法 FlatMap Window Aggregations Reduce
132 0
|
1月前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队,全球领先的流计算引擎缔造者,支撑双11万亿级数据处理,推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才,地点覆盖北京、杭州、上海。技术深度参与开源核心,打造企业级实时计算解决方案,助力全球企业实现毫秒洞察。
364 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。

热门文章

最新文章