查找相同的词(寻找相近的词有哪些)
766
2022-05-30
欢迎访问我的GitHub
这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos
本篇概览
本文是《Flink处理函数实战》系列的第三篇,上一篇《Flink处理函数实战之二:ProcessFunction类》学习了最简单的ProcessFunction类,今天要了解的KeyedProcessFunction,以及该类带来的一些特性;
关于KeyedProcessFunction
通过对比类图可以确定,KeyedProcessFunction和ProcessFunction并无直接关系:
KeyedProcessFunction用于处理KeyedStream的数据集合,相比ProcessFunction类,KeyedProcessFunction拥有更多特性,官方文档如下图红框,状态处理和定时器功能都是KeyedProcessFunction才有的:
介绍完毕,接下来通过实例来学习吧;
版本信息
开发环境操作系统:MacBook Pro 13寸, macOS Catalina 10.15.3
开发工具:IDEA ULTIMATE 2018.3
JDK:1.8.0_211
Maven:3.6.0
Flink:1.9.2
源码下载
如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blog_demos):
这个git项目中有多个文件夹,本章的应用在
flinkstudy
文件夹下,如下图红框所示:
实战简介
本次实战的目标是学习KeyedProcessFunction,内容如下:
监听本机9999端口,获取字符串;
将每个字符串用空格分隔,转成Tuple2实例,f0是分隔后的单词,f1等于1;
上述Tuple2实例用f0字段分区,得到KeyedStream;
KeyedSteam转入自定义KeyedProcessFunction处理;
自定义KeyedProcessFunction的作用,是记录每个单词最新一次出现的时间,然后建一个十秒的定时器,十秒后如果发现这个单词没有再次出现,就把这个单词和它出现的总次数发送到下游算子;
编码
继续使用《Flink处理函数实战之二:ProcessFunction类》一文中创建的工程flinkstudy;
创建bean类CountWithTimestamp,里面有三个字段,为了方便使用直接设为public:
package com.bolingcavalry.keyedprocessfunction; public class CountWithTimestamp { public String key; public long count; public long lastModified; }
创建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
最后是整个逻辑功能的主体: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
上述代码有几处需要重点关注的:
通过assignTimestampsAndWatermarks设置时间戳的时候,getCurrentWatermark返回null,因为用不上watermark;
processElement方法中,state.value()可以取得当前单词的状态,state.update(current)可以设置当前单词的状态,这个功能的详情请参考《深入了解ProcessFunction的状态操作(Flink-1.10)》;
registerProcessingTimeTimer方法设置了定时器的触发时间,注意这里的定时器是基于processTime,和官方demo中的eventTime是不同的;
定时器触发后,onTimer方法被执行,里面有这个定时器的全部信息,尤其是入参timestamp,这是原本设置的该定时器的触发时间;
验证
在控制台执行命令nc -l 9999,这样就可以从控制台向本机的9999端口发送字符串了;
在IDEA上直接执行ProcessTime类的main方法,程序运行就开始监听本机的9999端口了;
在前面的控制台输入aaa,然后回车,等待十秒后,IEDA的控制台输出以下信息,从结果可见符合预期:
继续输入aaa再回车,连续两次,中间间隔不要超过10秒,结果如下图,可见每一个Tuple2元素都有一个定时器,但是第二次输入的aaa,其定时器在出发前,aaa的最新出现时间就被第三次输入的操作给更新了,于是第二次输入aaa的定时器中的对比操作发现此时距aaa的最近一次(即第三次)出现还未达到10秒,所以第二个元素不会发射到下游算子:
下游算子收到的所有超时信息会打印出来,如下图红框,只打印了数量等于1和3的记录,等于2的时候因为在10秒内再次输入了aaa,因此没有超时接收,不会在下游打印:
至此,KeyedProcessFunction处理函数的学习就完成了,其状态读写和定时器操作都是很实用能力,希望本文可以给您提供参考;
欢迎关注华为云博客:程序员欣宸
学习路上,你不孤单,欣宸原创一路相伴…
Flink 控制台
版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。