Flink+Hologres亿级用户实时UV精确去重最佳实践

//此处使用csv文件作为数据源，也可以是kafka等DataStreamSourceodsStream=env.createInput(csvInput, typeInfo);
// 与维表join需要添加proctime字段，详见https://help.aliyun.com/document_detail/62506.htmlTableodsTable=tableEnv.fromDataStream(
odsStream,
$("uid"),
$("country"),
$("prov"),
$("city"),
$("ymd"),
$("proctime").proctime());
// 注册到catalog环境tableEnv.createTemporaryView("odsTable", odsTable);

// 创建Hologres维表，其中nsertIfNotExists表示查询不到则自行插入StringcreateUidMappingTable=String.format(
"create table uid_mapping_dim("+"  uid string,"+"  uid_int32 INT"+") with ("+"  'connector'='hologres',"+"  'dbname' = '%s',"//Hologres DB名+"  'tablename' = '%s',"//Hologres 表名+"  'username' = '%s',"//当前账号access id+"  'password' = '%s',"//当前账号access key+"  'endpoint' = '%s',"//Hologres endpoint+"  'insertifnotexists'='true'"+")",
database, dimTableName, username, password, endpoint);
tableEnv.executeSql(createUidMappingTable);
// 源表与维表joinStringodsJoinDim="SELECT ods.country, ods.prov, ods.city, ods.ymd, dim.uid_int32"+"  FROM odsTable AS ods JOIN uid_mapping_dim FOR SYSTEM_TIME AS OF ods.proctime AS dim"+"  ON ods.uid = dim.uid";
TablejoinRes=tableEnv.sqlQuery(odsJoinDim);

DataStream<Tuple6<String, String, String, String, Timestamp, byte[]>>processedSource=source// 筛选需要统计的维度（country, prov, city, ymd）    .keyBy(0, 1, 2, 3)
// 滚动时间窗口；此处由于使用读取csv模拟输入流，采用ProcessingTime，实际使用中可使用EventTime    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
// 触发器，可以在窗口未结束时获取聚合结果    .trigger(ContinuousProcessingTimeTrigger.of(Time.minutes(1)))
    .aggregate(
// 聚合函数，根据key By筛选的维度，进行聚合newAggregateFunction<Tuple5<String, String, String, String, Integer>,
RoaringBitmap,
RoaringBitmap>() {
@OverridepublicRoaringBitmapcreateAccumulator() {
returnnewRoaringBitmap();
            }
@OverridepublicRoaringBitmapadd(
Tuple5<String, String, String, String, Integer>in,
RoaringBitmapacc) {
// 将32位的uid添加到RoaringBitmap进行去重acc.add(in.f4);
returnacc;
            }
@OverridepublicRoaringBitmapgetResult(RoaringBitmapacc) {
returnacc;
            }
@OverridepublicRoaringBitmapmerge(
RoaringBitmapacc1, RoaringBitmapacc2) {
returnRoaringBitmap.or(acc1, acc2);
            }
     },
//窗口函数，输出聚合结果newWindowFunction<RoaringBitmap,
Tuple6<String, String, String, String, Timestamp, byte[]>,
Tuple,
TimeWindow>() {
@Overridepublicvoidapply(
Tuplekeys,
TimeWindowtimeWindow,
Iterable<RoaringBitmap>iterable,
Collector<Tuple6<String, String, String, String, Timestamp, byte[]>>out)
throwsException {
RoaringBitmapresult=iterable.iterator().next();
// 优化RoaringBitmapresult.runOptimize();
// 将RoaringBitmap转化为字节数组以存入Holo中byte[] byteArray=newbyte[result.serializedSizeInBytes()];
result.serialize(ByteBuffer.wrap(byteArray));
// 其中 Tuple6.f4(Timestamp) 字段表示以窗口长度为周期进行统计，以秒为单位out.collect(
newTuple6<>(
keys.getField(0),
keys.getField(1),
keys.getField(2),
keys.getField(3),
newTimestamp(
timeWindow.getEnd() /1000*1000),
byteArray));
        }
    });

// 计算结果转换为表TableresTable=tableEnv.fromDataStream(
processedSource,
$("country"),
$("prov"),
$("city"),
$("ymd"),
$("timest"),
$("uid32_bitmap"));
// 创建Hologres结果表, 其中Hologres的RoaringBitmap类型通过Byte数组存入StringcreateHologresTable=String.format(
"create table sink("+"  country string,"+"  prov string,"+"  city string,"+"  ymd string,"+"  timetz timestamp,"+"  uid32_bitmap BYTES"+") with ("+"  'connector'='hologres',"+"  'dbname' = '%s',"+"  'tablename' = '%s',"+"  'username' = '%s',"+"  'password' = '%s',"+"  'endpoint' = '%s',"+"  'connectionSize' = '%s',"+"  'mutatetype' = 'insertOrReplace'"+")",
database, dwsTableName, username, password, endpoint, connectionSize);
tableEnv.executeSql(createHologresTable);
// 写入计算结果到dws表tableEnv.executeSql("insert into sink select * from "+resTable);

--运行下面RB_AGG运算查询，可执行参数先关闭三阶段聚合开关（默认关闭）,性能更好sethg_experimental_enable_force_three_stage_agg=offSELECTcountry        ,prov        ,city        ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) ASuvFROMdws_appWHEREymd='20210329'GROUPBYcountry         ,prov         ,city;

--运行下面RB_AGG运算查询，可执行参数先关闭三阶段聚合开关（默认关闭）,性能更好sethg_experimental_enable_force_three_stage_agg=offSELECTcountry        ,prov        ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) ASuvFROMdws_appWHEREtime>'2021-04-19 18:00:00+08'andtime<'2021-04-19 19:00:00+08'GROUPBYcountry         ,prov;