Flink RoaringBitmap去重

1、RoaringBitmap的依赖

  <!-- 去重大哥-->
<dependency><groupId>org.roaringbitmap</groupId><artifactId>RoaringBitmap</artifactId><version>0.9.21</version>
</dependency>

2、Demo去重

package com.gwm.driver;import com.alibaba.fastjson.JSON;
import com.alibaba.flink.connectors.datahub.datastream.source.DatahubSourceFunction;
import com.aliyun.datahub.client.model.RecordEntry;
import com.gwm.pojo.EventSuccessInfo;
import com.gwm.utils.TimeToStampUtil;
import com.gwm.utils.getString;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
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.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import scala.Tuple4;import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.UUID;/*** @author yangyingchun* @version 1.0* @date 2022/11/14 16:26*/
public class EventOrderSuccessRoaringBitmap {private static String endPoint = "endPoint ";//private static String endPoint ="public endpoint";//公网访问（填写内网Endpoint，就不用填写公网Endpoint）。private static String projectName = "projectName ";private static String topicSourceName =  "topicSourceName ";
//    private static String topicSourceName =  "topicSourceName ";private static String accessId = "accessId ";private static String accessKey = "accessKey ";//设置消费的启动位点对应的时间。TimeToStampUtil.timeToStamp("2021-12-21") 此时间至少为当前时间
//    private static Long datahubStartInMs = TimeToStampUtil.timeToStamp("2023-02-23");private static Long datahubStartInMs = System.currentTimeMillis();private static Long datahubEndInMs=Long.MAX_VALUE;private static SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");private static SimpleDateFormat sd1 = new SimpleDateFormat("yyyy-MM-dd");private static Date startDate;static {try {startDate = sd1.parse(sd.format(new Date()));} catch (ParseException e) {e.printStackTrace();}};public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);env.enableCheckpointing(3600000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));env.setParallelism(1);DataStreamSource<List<RecordEntry>> aedata =  env.addSource(new DatahubSourceFunction(endPoint,projectName,topicSourceName,accessId,accessKey,datahubStartInMs,datahubEndInMs,20L,1000L,1000));DataStream<Tuple4<String, EventSuccessInfo, String, Long>> aecollectordataDataStream = aedata.flatMap(new FlatMapFunction<List<RecordEntry>, Tuple4<String, EventSuccessInfo, String, Long>>() {@Overridepublic void flatMap(List<RecordEntry> value, Collector<Tuple4<String, EventSuccessInfo, String, Long>> out) throws Exception {for (RecordEntry recordEntry : value) {String phone = getString.getString(recordEntry, "customer_phone");Long order_sn = Long.parseLong(getString.getString(recordEntry, "order_no"));String brand = getString.getString(recordEntry, "brand");String car_model = getString.getString(recordEntry, "car_model");String action_time = "null".equals(getString.getString(recordEntry, "paid_at"))||"".equals(getString.getString(recordEntry, "paid_at"))?null:sd.format(new Date(Long.parseLong(getString.getString(recordEntry, "paid_at"))/1000));Double paid_amount = "null".equals(getString.getString(recordEntry, "paid_amount"))?null:Double.parseDouble(getString.getString(recordEntry, "paid_amount"));String name = getString.getString(recordEntry, "customer_name");String operation_flag = getString.getString(recordEntry, "new_dts_sync_dts_after_flag");String order_time = "null".equals(getString.getString(recordEntry, "order_time"))||"".equals(getString.getString(recordEntry, "order_time"))?null:sd.format(new Date(Long.parseLong(getString.getString(recordEntry, "order_time"))/1000));String order_state = getString.getString(recordEntry, "order_state"); //'订购成功'Date add_time ="null".equals(getString.getString(recordEntry, "order_time"))||"".equals(getString.getString(recordEntry, "order_time"))?null:new Date(Long.parseLong(getString.getString(recordEntry, "order_time")) / 1000);
//                    startDate = sd1.parse(sd.format(new Date()));System.out.println(order_state+"====startDate:"+startDate+"====paid_at:"+order_time+"=====phone+order_sn:"+phone+"--"+order_sn);//这里有三个问题，// 1、技术+业务：因为获取的是数据库操作日志，所以数据是重复的，（已经做了重复校验，确保不会重复发且无时效性）// 2、技术：如果操作了历史数据，且用户的订单状态恰好还是订购成功时，也会触达，是不是要加限制，加的话加什么合适，//    新增且当天（很多数据是获取不到时间的）？还是所有时间都推，再ma测加一个时间的控制条件//    结论：空的也要，// 3、业务：需要明确订购成功的规则，否则极易造成异常, order_state=12当前是订购成功 能复用吗if (
//                            "12".equals(order_state)&&"Y".equals(operation_flag)
//                             &&!StringUtils.isNullOrWhitespaceOnly(order_time)
//                             &&add_time.after(startDate)){EventSuccessInfo eventSuccessInfo = new EventSuccessInfo(phone, order_sn, brand, car_model, action_time, paid_amount, name, operation_flag,order_time,order_state);//                    System.out.println(eventSuccessInfo);Tuple4<String, EventSuccessInfo, String, Long> tuple4= new Tuple4<String, EventSuccessInfo, String, Long>("test_event_order_success",eventSuccessInfo,UUID.randomUUID().toString().replace("-",""),System.currentTimeMillis());out.collect(tuple4);}}}});KeyedStream<Tuple4<String, EventSuccessInfo, String, Long>, String> tuple4StringKeyedStream= aecollectordataDataStream.keyBy(x -> x._2().getPhone());//        StateTtlConfig ttlConfig = StateTtlConfig
//                .newBuilder(Time.days(2))
//                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
//                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
//                .build();//create StateDescriptor//这里进行状态注册通过bitmap高效存储实现去重，当然bitmap去重只适合bigint场景ValueStateDescriptor<Roaring64Bitmap> bitmapDescriptor = new ValueStateDescriptor("Roaring64Bitmap",TypeInformation.of(new TypeHint<Roaring64Bitmap>() {}));//手机号去重逻辑 通过Roaring64BitmapSingleOutputStreamOperator<Tuple4<String, EventSuccessInfo, String, Long>> map = tuple4StringKeyedStream.filter(new RichFilterFunction<Tuple4<String, EventSuccessInfo, String, Long>>() {//1.定义状态 进行手机号去重private transient ValueState<Roaring64Bitmap> bitmapState;@Overridepublic void open(Configuration parameters) throws Exception {// 设置状态生命周期
//                StateTtlConfig stateTtlConfig = new StateTtlConfig
//                        .Builder(Time.days(1)) // 周期为1天
//                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 创建或者更新状态时重新刷新生命周期
//                        .build();bitmapState = getRuntimeContext().getState(bitmapDescriptor);;}@Overridepublic boolean filter(Tuple4<String, EventSuccessInfo, String, Long> value) throws Exception {//由于本程序只筛选订购成功的，所以每个手机号+每个订单唯一确认一条数据（订单状态已经在上游过滤过了）Roaring64Bitmap bitmap = bitmapState.value();if (bitmap == null) {bitmap = new Roaring64Bitmap();}if (!bitmap.contains(value._2().getOrder_sn())) {bitmap.addLong(value._2().getOrder_sn());bitmapState.update(bitmap);return true;}return false;}});//因为是binlog，但需求只要数据时间是当天的 ：通过flink定时器 定义每天零晨更新比较时间SingleOutputStreamOperator<Tuple4<String, EventSuccessInfo, String, Long>> process = map.keyBy(x -> x._2().getPhone()).process(new KeyedProcessFunction<String, Tuple4<String, EventSuccessInfo, String, Long>, Tuple4<String, EventSuccessInfo, String, Long>>() {//1.定义状态 进行手机号去重private ValueState<String> timeSate;@Overridepublic void processElement(Tuple4<String, EventSuccessInfo, String, Long> value, Context ctx, Collector<Tuple4<String, EventSuccessInfo, String, Long>> out) throws Exception {//获取格林威治标准时间的第二天00:00:00即获取北京时间的第二天08:00:00
//                long ts = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (1000 * 60 * 60 * 24);//获取北京时间的第二天00:00:00long ts = ( ctx.timerService().currentProcessingTime()/(1000*60*60*24) + 1) * (1000*60*60*24)- 8 * 60 * 60 * 1000;//                long ts = 1677054000000L;//如果注册相同数据的TimeTimer，后面的会将前面的覆盖，即相同的timeTimer只会触发一次ctx.timerService().registerProcessingTimeTimer(ts);out.collect(value);}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple4<String, EventSuccessInfo, String, Long>> out) throws Exception {//定时器质性，每天凌晨更新开始时间
//                System.out.println(timestamp);System.out.println("定时器执行了：" + timestamp);//状态初始化timeSate.clear();startDate = sd1.parse(sd.format(new Date()));System.out.println(startDate);
//                startDate = sd1.parse("2023-02-01");}});SingleOutputStreamOperator<Tuple4<String, String, String, Long>> jsonString = process.map(new MapFunction<Tuple4<String, EventSuccessInfo, String, Long>, Tuple4<String, String, String, Long>>() {@Overridepublic Tuple4<String, String, String, Long> map(Tuple4<String, EventSuccessInfo, String, Long> value) throws Exception {return new Tuple4<String, String, String, Long>(value._1(),JSON.toJSONString(value._2()),value._3(),value._4());}});jsonString.print();
//        jsonString.addSink(new EventOmsSuccessSink());env.execute("EventOrderSuccess===>");}
}

3、注意：Roaring64Bitmap 去重只适合去重整形情况

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.xdnf.cn/news/144540.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

Android 使用kotlin+注解+反射+泛型实现MVP架构

Android 使用kotlin+注解+反射+泛型实现MVP架构

一&#xff0c;MVP模式的定义 ①Model&#xff1a;用于存储数据。它负责处理领域逻辑以及与数据库或网络层的通信。 ②View&#xff1a;UI层&#xff0c;提供数据可视化界面&#xff0c;并跟踪用户的操作&#xff0c;以便通知presenter。 ③Presenter&#xff1a;从Model层获…
阅读更多...
部署Kafka

部署Kafka

kafka&#xff1a;kafka_2.13-3.5.1 NOTE: Your local environment must have Java 8 installed. Apache Kafka can be started using ZooKeeper or KRaft. To get started with either configuration follow one the sections below but not both. 1 Windows单机 1.1 Kafka w…
阅读更多...
【SpringBoot实践】事务和事务传播机制失效原因正确使用事务的建议

【SpringBoot实践】事务和事务传播机制失效原因正确使用事务的建议

文章目录 1.概述2.事务与事务传播2.1 声明式事务说明2.2.声明式事务失效原因2.3.事务的传播机制2.4.事务传播失效原因 3.事务使用建议4.总结 1.概述 我们在开发工作中经常会使用到事务&#xff0c;来保证数据库做增、删、改操作时的数据一致性&#xff0c;在使用Spring来处理事…
阅读更多...
2023智慧云打印小程序源码多店铺开源版 +前端

2023智慧云打印小程序源码多店铺开源版 +前端

智慧自助云打印系统/智慧云打印小程序源码 前端 这是一款全新的基于Thinkphp的最新自助打印系统&#xff0c;最新UI界面设计的云打印小程序源码
阅读更多...
单目标应用：基于螳螂搜索算法（Mantis Search Algorithm，MSA）的微电网优化调度MATLAB

单目标应用：基于螳螂搜索算法（Mantis Search Algorithm，MSA）的微电网优化调度MATLAB

一、螳螂搜索算法 螳螂搜索算法&#xff08;Mantis Search Algorithm&#xff0c;MSA&#xff09;由Mohamed Abdel-Basset等人于2023年提出&#xff0c;该算法模拟螳螂独特的狩猎和性同类相食行为。MSA由三个优化阶段组成&#xff0c;包括寻找猎物&#xff08;探索&#xff09…
阅读更多...
【Unity3D赛车游戏制作】开始界面场景搭建

【Unity3D赛车游戏制作】开始界面场景搭建

&#x1f468;‍&#x1f4bb;个人主页&#xff1a;元宇宙-秩沅 &#x1f468;‍&#x1f4bb; hallo 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! &#x1f468;‍&#x1f4bb; 本文由 秩沅 原创 &#x1f468;‍&#x1f4bb; 收录于专栏&#xff1a;Uni…
阅读更多...
13.（开发工具篇github）如何在GitHub上上传本地项目

13.（开发工具篇github）如何在GitHub上上传本地项目

一：创建GitHub账户并安装Git 二：创建一个新的仓库（repository） 三、拉取代码 git clone https://github.com/ainier-max/myboot.git git clone git@github.com:ainier-max/myboot.git四、拷贝代码到拉取后的工程 五、上传代码 （1）添加所有文件到暂存
阅读更多...
Idea引入thymeleaf失败解决方法

Idea引入thymeleaf失败解决方法

报错 Whitelabel Error Page This application has no explicit mapping for /error, so you are seeing this as a fallback.Fri Sep 29 09:42:00 CST 2023 There was an unexpected error (typeNot Found, status404). 原因&#xff1a;html没有使用thymeleaf 首先要引入…
阅读更多...
14.（开发工具篇github）如何在Github配置ssh key

14.（开发工具篇github）如何在Github配置ssh key

第一步&#xff1a;检查本地主机是否已经存在ssh key 上图表示已存在。跳第三步 第二步&#xff1a;生成ssh key ssh-keygen -t rsa -C "xxxxxx.com"第三步&#xff1a;获取ssh key公钥内容&#xff08;id_rsa.pub&#xff09; cat id_rsa.pub第四步&#xff1a;G…
阅读更多...
【开发篇】十二、缓存框架JetCache

【开发篇】十二、缓存框架JetCache

文章目录 0、介绍1、JetCache远程缓存2、JetCache本地缓存3、标准配置文件4、JetCache方法缓存注解--Cached5、Cached4、CacheUpdate5、CacheInvalidate6、CacheRefresh7、缓存统计报告 上篇完成了Spring Cache底层技术的各种切换&#xff0c;但各个技术有各自的优缺点&#xf…
阅读更多...
UE5 ChaosVehicles载具研究

UE5 ChaosVehicles载具研究

一、基本组成 载具Actor类名称&#xff1a;WheeledVehiclePawn Actor最原始的结构 官方增加了两个摇臂相机&#xff0c;可以像驾驶游戏那样切换多机位、旋转观察 选择骨骼网格体、动画蓝图类、开启物理模拟 二、SportsCar_Pawn 角阻尼&#xff1a;物体旋转的阻力。数值越大…
阅读更多...
3D孪生场景搭建：模型阵列摆放

3D孪生场景搭建：模型阵列摆放

阵列摆放概念 阵列摆放是指将物体、设备或元件按照一定的规则和间距排列组合的方式。在工程和科学领域中&#xff0c;阵列式摆放常常用于优化空间利用、提高效率或增强性能。 阵列摆放通常需要考虑间距、角度、方向、对称性等因素&#xff0c;以满足特定的要求和设计目标。不同…
阅读更多...
同城信息服务源码 本地生活服务小程序源码

同城信息服务源码 本地生活服务小程序源码

同城信息服务源码 本地生活服务小程序源码 功能介绍&#xff1a; 基本设置&#xff1a;网站参数、安全设置、分站管理、支付设置、操作日志、地区设置、公交地铁、国际区号、清理缓存、模板风格、模块管理、域名管理、底部菜单、消息通知、登录设置 其他设置&#xff1a;关键…
阅读更多...
Python 基于 Yolov8 + CPU 实现物体检测

Python 基于 Yolov8 + CPU 实现物体检测

目录 一、开发环境 二、安装 Python 基于 Yolov8 物体检测关联库 2.1 打开命令提示符&#xff08;cmd&#xff09;或终端&#xff0c;安装库 2.2 关联库安装过程遇到的问题 三、基于 Yolov8 物体检测代码实现&#xff08;完整&#xff09; 3.1 Yolov8 物体检测完整代码…
阅读更多...
力扣：112. 路径总和（Python3）

力扣：112. 路径总和（Python3）

题目&#xff1a; 给你二叉树的根节点 root 和一个表示目标和的整数 targetSum 。判断该树中是否存在 根节点到叶子节点 的路径&#xff0c;这条路径上所有节点值相加等于目标和 targetSum 。如果存在&#xff0c;返回 true &#xff1b;否则&#xff0c;返回 false 。 叶子节点…
阅读更多...
Flink状态

Flink状态

8.1 Flink中的状态 8.1.1 概述 状态的分类 1&#xff09;托管状态&#xff08;Managed State&#xff09;和原始状态&#xff08;Raw State&#xff09; Flink的状态有两种&#xff1a;托管状态&#xff08;Managed State&#xff09;和原始状态&#xff08;Raw State&#…
阅读更多...
PICO首届XR开发者挑战赛正式启动，助推行业迈入“VR+MR”新阶段

PICO首届XR开发者挑战赛正式启动，助推行业迈入“VR+MR”新阶段

9月25日&#xff0c;“PICO 2023首届XR开发者挑战赛”&#xff08;下文简称“挑战赛”&#xff09;媒体启动会在北京圆满落幕&#xff0c;官方赛事报名通道已于今日开启。据悉&#xff0c;本次挑战赛是PICO首次针对全球开发者举办的大型挑战赛事&#xff0c;旨在与开发者保持连…
阅读更多...
redis介绍

redis介绍

一、简介 Redis 与其他 key - value 缓存产品有以下三个特点&#xff1a; Redis支持数据的持久化&#xff0c;可以将内存中的数据保存在磁盘中&#xff0c;重启的时候可以再次加载进行使用。 Redis不仅仅支持简单的key-value类型的数据&#xff0c;同时还提供list&#xff0c;…
阅读更多...
从入门到精通：详解SVN版本控制系统的使用方法

从入门到精通：详解SVN版本控制系统的使用方法

从入门到精通&#xff1a;详解SVN版本控制系统的使用方法 一、引言1.1、版本控制的概念和重要性1.2、流行的版本控制系统SVN 二、SVN基础知识2.1、SVN的基本概念和术语解释2.2、SVN的工作原理和架构 三、创建SVN仓库3.1、创建本地仓库3.2、配置访问权限 四、使用SVN进行版本控制…
阅读更多...
基于视频技术与AI检测算法的体育场馆远程视频智能化监控方案

基于视频技术与AI检测算法的体育场馆远程视频智能化监控方案

一、方案背景 近年来&#xff0c;随着居民体育运动意识的增强&#xff0c;体育场馆成为居民体育锻炼的重要场所。但使用场馆内的器材时&#xff0c;可能发生受伤意外&#xff0c;甚至牵扯责任赔偿纠纷问题。同时&#xff0c;物品丢失、人力巡逻成本问题突出&#xff0c;体育场…
阅读更多...
最新文章

Copyright @ 2022~2023