Flink算子distinct

对于flink的stream没有原生的distinct方法,下面使用ValueState实现一个:

SingleOutputStreamOperator<PO> newStream = stream.keyBy(new KeySelector<PO, PO>() {@Overridepublic PO getKey(PO value) throws Exception {return value;}}).process(new KeyedProcessFunction<PO, PO, PO>() {private ValueState<Integer> state;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);ValueStateDescriptor<Integer> desctiptor = new ValueStateDescriptor<>("distinct", Integer.class);state=getRuntimeContext().getState(desctiptor);}@Overridepublic void processElement(PO value, KeyedProcessFunction<PO, PO, PO>.Context ctx, Collector<PO> out) throws Exception {if(state.value()==null) {state.update(1);out.collect(value);}}});

Flink算子distinct相关推荐

udp怎么保证不丢包_在 Flink 算子中使用多线程如何保证不丢数据？
分析痛点笔者线上有一个 Flink 任务消费 Kafka 数据,将数据转换后,在 Flink 的 Sink 算子内部调用第三方 api 将数据上报到第三方的数据分析平台.这里使用批量同步 api,即 ...
线程中如何使用对象_在 Flink 算子中使用多线程如何保证不丢数据？
简介: 本人通过分析痛点.同步批量请求优化为异步请求.多线程 Client 模式.Flink 算子内多线程实现以及总结四部分帮助大家理解 Flink 中使用多线程的优化及在 Flink 算子中使用多线 ...
Flink 算子Function实例化的坑
问题回顾关于一段代码: object MySingleObj{// 陷阱:// 单例对象中一个是可变引用,一个是可变数组var str:String = _val list = new ListBu ...
Flink系列之：Java代码实现深入浅出的理解Flink算子的使用方法
Flink系列之:Java代码实现深入浅出的理解Flink算子的使用方法一.Map算子二.filter算子三.flatMap算子四.keyBy算子五.Reduce算子六.union算子七 ...
Flink算子(Filter、KeyBy、Reduce和Aggregate)
Filter算子:过滤作用 filter算子过滤函数过滤函数,过滤出需要的数据,对传入的数据进行判断,如果返回true则该元素继续向下传递,如果返回false则该元素将被过滤掉.比如:如果返回来的价 ...
Flink算子(ProcessFunction，map和Flatmap)
Flink提供三层API,每个API在简洁性和表达之间提供不同的权衡,并针对不同的用例 SQL/Table API(dynamic tables) DataStream API(streams,win ...
Flink 算子状态与键控状态总结
Flink 常用状态算子状态(Operatior State) 键控状态(Keyed State) 状态后端(State Backends) Flink 中的状态如下图所示,为一条数据流经过fli ...
Flink 算子Operators总结
Operator 作用流的转换 map 将一个元素转换成另外一个元素 DataStream → DataStream本 flapmap 将几个的一个元素转换为零个,一个或者多个 DataStream ...
【2】flink数据流转换算子
[README] 本文记录了flink对数据的转换操作,包括基本转换,map,flatMap,filter: 滚动聚合(min minBy max maxBy sum): 规约聚合-reduce: ...

Flink算子distinct

Flink算子distinct相关推荐

最新文章

热门文章