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

Stella981
• 阅读 398

欢迎访问我的GitHub

https://github.com/zq2599/blog\_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

Flink处理函数实战系列链接

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

本篇概览

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

关于KeyedProcessFunction

通过对比类图可以确定,KeyedProcessFunction和ProcessFunction并无直接关系:
Flink处理函数实战之三:KeyedProcessFunction类
KeyedProcessFunction用于处理KeyedStream的数据集合,相比ProcessFunction类,KeyedProcessFunction拥有更多特性,官方文档如下图红框,状态处理和定时器功能都是KeyedProcessFunction才有的:
Flink处理函数实战之三: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文件夹下,如下图红框所示:
Flink处理函数实战之三:KeyedProcessFunction类

实战简介

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

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

编码

  1. 继续使用《Flink处理函数实战之二:ProcessFunction类》一文中创建的工程flinkstudy;

  2. 创建bean类CountWithTimestamp,里面有三个字段,为了方便使用直接设为public:

    package com.bolingcavalry.keyedprocessfunction;public class CountWithTimestamp { public String key; public long count; public long lastModified;}

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

    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)); } }}

  4. 最后是整个逻辑功能的主体:ProcessTime.java,这里面有自定义的KeyedProcessFunction子类,还有程序入口的main方法,代码在下面列出来之后,还会对关键部分做介绍:

    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<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> { // 自定义状态 private ValueState state; @Override public void open(Configuration parameters) throws Exception { // 初始化状态,name是myState state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class)); } @Override public void proces.........

点赞
收藏
评论区
推荐文章
秃头王路飞 秃头王路飞
4个月前
webpack5手撸vue2脚手架
webpack5手撸vue相信工作个12年的小伙伴们在面试的时候多多少少怕被问到关于webpack方面的知识,本菜鸟最近闲来无事,就尝试了手撸了下vue2的脚手架,第一次发帖实在是没有经验,望海涵。languageJavaScript"name":"vuecliversion2","version":"1.0.0","desc
光头强的博客 光头强的博客
4个月前
Java面向对象试题
1、请创建一个Animal动物类,要求有方法eat()方法,方法输出一条语句“吃东西”。创建一个接口A,接口里有一个抽象方法fly()。创建一个Bird类继承Animal类并实现接口A里的方法输出一条有语句“鸟儿飞翔”,重写eat()方法输出一条语句“鸟儿吃虫”。在Test类中向上转型创建b对象,调用eat方法。然后向下转型调用eat()方
刚刚好 刚刚好
4个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
小森森 小森森
4个月前
校园表白墙微信小程序V1.0 SayLove -基于微信云开发-一键快速搭建,开箱即用
后续会继续更新,敬请期待2.0全新版本欢迎添加左边的微信一起探讨!项目地址:(https://www.aliyun.com/activity/daily/bestoffer?userCodesskuuw5n)\2.Bug修复更新日历2.情侣脸功能大家不要使用了,现在阿里云的接口已经要收费了(土豪请随意),\\和注意
晴空闲云 晴空闲云
4个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
3个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
Stella981 Stella981
1年前
Flink的sink实战之二:kafka
欢迎访问我的GitHubhttps://github.com/zq2599/blog\_demos内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;本篇概览本文是《Flink的sink实战》系列的第二篇,前文《Flink的sink实战之一:初探》对sink有了基本的了解,本
Stella981 Stella981
1年前
Flink on Yarn三部曲之三:提交Flink任务
欢迎访问我的GitHubhttps://github.com/zq2599/blog\_demos内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;本文是《FlinkonYarn三部曲》系列的终篇,先简单回顾前面的内容:1.《FlinkonYarn三部曲之一:准备工作》
helloworld_28799839 helloworld_28799839
4个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue