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

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 使用实时计算Flink版 + Hologres 方式,并基于 RoaringBitmap,实时对用户标签去重。

GitHub 地址
https://github.com/apache/flink
欢迎大家给 Flink 点赞送 star~

1. 背景信息

在用户行为分析和圈人场景中,经常需要从亿级甚至几十亿级用户中快速筛选出符合特定标签的指标结果。UV 便是行为分析中最常见的指标,代表访问网页的自然人,可以引申为某段时间内某个指标精确去重后的量。例如大促时,电商商家需要实时计算店铺的实时 UV,并根据 UV 情况及时调整运营策略,从而达成销售目标。

在计算用户 UV 时,由于业务需求不同,计算的维度和数据量也不同,通常来讲,会有以下几点诉求:

  • 用户数据量大,每天几亿条,维度多(10+ 以上),需要支持各维度间任意组合查询;
  • 查询时间需要更灵活,不仅局限于天、周、月、年等,还需要支持更细粒度实时更新查询;
  • 需要对用户数精确去重。

面对上诉高复杂度 UV 计算场景,业界常见的手段包括使用 Apche Kylin 等预计算系统或者 Flink + MySQL 的固定维度组合方案,但也会遇见以下几个痛点:

  • 需求维度过多时,会带来存储爆炸,预计算时间长;
  • 精确去重需要消耗大量资源,容易 OOM;
  • 实时更新难,无法支持更加灵活开放的时间周期处理。

Hologres 是基于分析服务一体化理念(Hybrid Serving & Analytical Processing)设计的实时数仓产品,它采用分布式架构,支持数据实时写入,高并发、低延时的分析处理 PB 级数据,兼容 PostgreSQL 协议,使用最熟悉的工具就能进行开发。

Hologres 与实时计算Flink版有着强大的融合优化,支持 Flink 数据高通量实时写入,写入即可见;支持 Flink SQL 维表关联,以及作为 CDC Source 事件驱动开发;同时,Hologres 也支持 RoaringBitmap,利用其超高算法能力和高效 Bitmap 压缩能力,对用户标签筛选,对去重等场景有着高性能的支持。

在上诉 UV 计算场景中,可以使用实时计算Flink版 + Hologres 方式,并基于 RoaringBitmap,实时对用户标签去重。这样的方式,可以较细粒度的实时得到用户 UV、PV 数据,同时便于根据需求调整最小统计窗口(如最近 5 分钟的 UV),实现类似实时监控的效果,更好的在大屏等 BI 展示。相较于以天、周、月等单位的去重,更适合在活动日期进行更细粒度的统计,并且通过简单的聚合,也可以得到较大时间单位的统计结果。

该方案数据链路简单,可以任意维度灵活计算,只需要一份 Bitmap 存储,也没有存储爆炸的问题,还能保证实时更新,从而实现更实时、开发更灵活、功能更完善的多维分析数仓。

下面将会就 UV 计算场景讲解具体操作步骤。

2. 业务架构图

  1. 实时计算Flink版实时订阅实时采集的数据,数据源可以来源于日志数据,如 Kafka 等;
  2. 实时计算Flink版对数据做进一步加工处理;
  3. Hologres 对实时计算Flink版实时写入的数据实时查询;
  4. 最终查询的数据对接可视化工具,如 DataV 等,用作大屏展示。

img

3. 详细业务流程

  1. 实时计算Flink版将流式数据转化为表与维表进行 JOIN 操作,再转化为流式数据。此举可以利用 Hologres 维表的 insertIfNotExists 特性结合自增字段实现高效的 uid 映射;
  2. 实时计算Flink版把关联的结果数据按照时间窗口进行处理,根据查询维度使用 RoaringBitmap 进行聚合,并将查询维度以及聚合的 uid 存放在聚合结果表,其中聚合出的 uid 结果放入 Hologres 的 RoaringBitmap 类型的字段中;
  3. 查询时,与离线方式相似,直接按照查询条件查询聚合结果表,并对其中关键的 RoaringBitmap 字段做 or 运算后并统计基数,即可得出对应用户数;
  4. 具体数据处理流程如下图所示:
    img

4. 方案最佳实践

4.1 前提条件

4.2 操作步骤

4.2.1 Hologres 创建相关基础表

1)在 Hologres 创建表 uid_mapping 为 uid 映射表,用于映射 uid 到 32位 int 类型。如果原始 uid 已经是 int32 类型,此步骤可忽略。

  • 常见的业务系统或者埋点中的用户 ID 很多是字符串类型或 Long 类型,因此需要使用 uid_mapping 类型构建一张映射表。 RoaringBitmap 类型要求用户 ID 必须是 32位 int 类型,而且越稠密越好(即用户 ID 最好连续)。映射表利用 Hologres 的 SERIAL 类型(自增的 32位 int)来实现用户映射的自动管理和稳定映射。
  • 由于是实时数据, 在 Hologres 中该表为行存表,以提高 Flink 维表实时 JOIN 的 QPS。
BEGIN;
CREATE TABLE public.uid_mapping (
uid text NOT NULL,
uid_int32 serial,
PRIMARY KEY (uid)
);
--将uid设为clustering_key和distribution_key便于快速查找其对应的int32值
CALL set_table_property('public.uid_mapping', 'clustering_key', 'uid');
CALL set_table_property('public.uid_mapping', 'distribution_key', 'uid');
CALL set_table_property('public.uid_mapping', 'orientation', 'row');
COMMIT;

2)创建表 dws_app 为基础聚合表,用于存放在基础维度上聚合后的结果。

  • 使用 RoaringBitmap 前需要创建 RoaringBitmap extention,同时也需要 Hologres 实例为 0.10 及以上版本。
CREATE EXTENSION IF NOT EXISTS roaringbitmap;
  • 为了更好的性能,建议根据基础聚合表数据量合理的设置 Shard 数,但建议基础聚合表的 Shard 数设置不超过计算资源总 Core 数的 60%。推荐使用以下哨兵表方式,设置 Table Group 的 Shard 数。
--新建shard数为16的Table Group,
--本次测试数据量百万级,总计算资源为128core,设置shard数为16
BEGIN;
CREATE TABLE tg16 (a int);                             --Table Group哨兵表
CALL set_table_property('tg16', 'shard_count', '16'); 
COMMIT;
  • 相比离线结果表,此结果表增加了时间戳字段,用于实现以 Flink 窗口周期为单位的统计。结果表 DDL 如下:
BEGIN;
CREATE TABLE dws_app(
  country text,
  prov text,
  city text, 
  ymd text NOT NULL,  --日期字段
  timetz TIMESTAMPTZ,  --统计时间戳,可以实现以Flink窗口周期为单位的统计
  uid32_bitmap roaringbitmap, -- 使用roaringbitmap记录uv
 PRIMARY KEY (country, prov, city, ymd, timetz)--查询维度和时间作为主键,防止重复插入数据
);
CALL set_table_property('public.dws_app', 'orientation', 'column');
--日期字段设为clustering_key和event_time_column,便于过滤
CALL set_table_property('public.dws_app', 'clustering_key', 'ymd');
CALL set_table_property('public.dws_app', 'event_time_column', 'ymd');
--等价于将表放在shard数为16的table group
call set_table_property('public.dws_app', 'colocate_with', 'tg16');
--group by字段设为distribution_key
CALL set_table_property('public.dws_app', 'distribution_key', 'country,prov,city');
COMMIT;

4.2.2 实时计算Flink版实时读取数据并更新 dws_app 基础聚合表

实时计算Flink版中的完整示例源码请见alibabacloud-hologres-connectors examples,下面是在实时计算Flink版中的具体操作步骤:

  1. Flink 流式读取数据源(DataStream),并转化为源表(Table)。

    在 Flink 中使用流式读取数据源,数据源可以是 CSV 文件,也可以来源于 Kafka、Redis 等,可以根据业务场景准备,此次不再具体展开讲述。在 Flink 中转化为源表的代码示例如下:

//此处使用csv文件作为数据源,也可以是kafka/redis等
DataStreamSource odsStream = env.createInput(csvInput, typeInfo);
// 与维表join需要添加proctime字段,详见https://help.aliyun.com/document_detail/62506.html
Table odsTable =
    tableEnv.fromDataStream(
    odsStream,
    $("uid"),
    $("country"),
    $("prov"),
    $("city"),
    $("ymd"),
    $("proctime").proctime());
// 注册到catalog环境
tableEnv.createTemporaryView("odsTable", odsTable);
  1. 在 Flink 中将源表与 Hologres 维表(uid_mapping)进行关联。

    • 在 Flink 中创建 Hologres 维表,需要使用 insertIfNotExists 参数,即查询不到数据时自行插入,uid_int32 字段便可以利用 Hologres 的 Serial 类型自增创建。
    • 将 Flink 源表与 Hologres 维表进行关联(JOIN)
// 创建Hologres维表,其中insertIfNotExists表示查询不到则自行插入
String createUidMappingTable =
    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);

// 源表与维表join
String odsJoinDim =
    "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";
Table joinRes = tableEnv.sqlQuery(odsJoinDim);
  1. 将关联结果转化为 DataStream,通过 Flink 时间窗口处理,结合 RoaringBitmap 进行对指标进行去重处理。
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筛选的维度,进行聚合
    new AggregateFunction<
        Tuple5<String, String, String, String, Integer>,
        RoaringBitmap,
        RoaringBitmap>() {
            @Override
            public RoaringBitmap createAccumulator() {
                return new RoaringBitmap();
            }

            @Override
            public RoaringBitmap add(
                Tuple5<String, String, String, String, Integer> in,
                RoaringBitmap acc) {
                // 将32位的uid添加到RoaringBitmap进行去重
                acc.add(in.f4);
                return acc;
            }

            @Override
            public RoaringBitmap getResult(RoaringBitmap acc) {
                return acc;
            }

            @Override
            public RoaringBitmap merge(
                RoaringBitmap acc1, RoaringBitmap acc2) {
                return RoaringBitmap.or(acc1, acc2);
            }
        },
    //窗口函数,输出聚合结果
    new WindowFunction<
        RoaringBitmap,
        Tuple6<String, String, String, String, Timestamp, byte[]>,
        Tuple,
        TimeWindow>() {
            @Override
            public void apply(
                Tuple keys,
                TimeWindow timeWindow,
                Iterable<RoaringBitmap> iterable,
                Collector<
                Tuple6<String, String, String, String, Timestamp, byte[]>> out)
                throws Exception {
                RoaringBitmap result = iterable.iterator().next();

                // 优化RoaringBitmap
                result.runOptimize();
                // 将RoaringBitmap转化为字节数组以存入Holo中
                byte[] byteArray = new byte[result.serializedSizeInBytes()];
                result.serialize(ByteBuffer.wrap(byteArray));

                // 其中 Tuple6.f4(Timestamp) 字段表示以窗口长度为周期进行统计,以秒为单位
                out.collect(
                    new Tuple6<>(
                        keys.getField(0),
                        keys.getField(1),
                        keys.getField(2),
                        keys.getField(3),
                        new Timestamp(
                            timeWindow.getEnd() / 1000 * 1000),
                        byteArray));
        }
    });
  1. 写入 Hologres 结果表。

经过 Flink 去重处理的数据写入至 Hologres 结果表 dws_app,但需要注意的是,Hologres 中 RoaringBitmap 类型在 Flink 中对应 Byte 数组类型。Flink 中代码如下:

 // 计算结果转换为表
Table resTable =
    tableEnv.fromDataStream(
        processedSource,
        $("country"),
        $("prov"),
        $("city"),
        $("ymd"),
        $("timest"),
        $("uid32_bitmap"));

// 创建Hologres结果表, 其中Hologres的RoaringBitmap类型通过Byte数组存入
String createHologresTable =
    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);

4.2.3 数据查询

在 Hologres 中对结果表(dws_app)进行 UV 计算。按照查询维度做聚合计算,查询 Bitmap 基数,得出 Group by 条件下的用户数

  • 示例查询 1:查询某天内各个城市的 uv。
--运行下面RB_AGG运算查询,可执行参数先关闭三阶段聚合开关(默认关闭),性能更好,此步骤可选
set hg_experimental_enable_force_three_stage_agg=off   

SELECT  country
        ,prov
        ,city
        ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) AS uv
FROM    dws_app
WHERE   ymd = '20210329'
GROUP BY country
         ,prov
         ,city
;
  • 示例查询 2:查询某段时间内各个省份的 UV、PV。
--运行下面RB_AGG运算查询,可执行参数先关闭三阶段聚合开关(默认关闭),性能更好,此步骤可选
set hg_experimental_enable_force_three_stage_agg=off   

SELECT  country
        ,prov
        ,RB_CARDINALITY(RB_OR_AGG(uid32_bitmap)) AS uv
        ,SUM(1) AS pv
FROM    dws_app
WHERE   time > '2021-04-19 18:00:00+08' and time < '2021-04-19 19:00:00+08'
GROUP BY country
         ,prov
;

4.2.4 可视化展示

计算出 UV、PV 和大多数情况需要用 BI 工具以更直观的方式可视化展示,由于需要使用 RB_CARDINALITY 和 RB_OR_AGG 进行聚合计算,需要使用 BI 的自定义聚合函数的能力,常见的具备该能力的 BI 包括 Apache Superset 和 Tableau,下面将会讲述这两个 BI 工具的最佳实践。

使用 Apache Superset

Apache Superset 对接 Hologres 的方式,请参考产品手册。在 Superset 中可以直接使用 dws_app 表作为 Dataset 使用

img

并且在数据集中,创建一个单独 Metrics,名为 UV,表达式如下:

RB_CARDINALITY(RB_OR_AGG(uid32_bitmap))。

img

然后就可以开始探索数据了。

img

img

当然也可以创建 Dashborad。

img

使用 Tableau

Tableau 对接 Hologres 的方式,请参考产品手册。可以使用 Tableau 的直通函数直接实现自定义函数的能力,详细介绍请参照Tableau 的手册。在 Tableau 对接 Hologres 后,可以创建一个计算字段,表达式如下:

RAWSQLAGG_INT("RB_CARDINALITY(RB_OR_AGG(%1))", [Uid32 Bitmap])。

img

然后就可以开始探索数据了。

img

当然也可以创建 Dashborad。

img

5. 总结

能支撑亿级用户的实时 UV 计算,是充分利用了实时计算Flink版和 Hologres 各自的优秀能力:

  • 实时计算Flink版的实时事件驱动能力、丰富的窗口定义能力、可扩展的编程接口,支持了 RoaringBitmap 累加行为的定义;
  • Hologres 实时更新能力、实时写入能力,保证了数据的高吞吐写入和生效;
  • Hologres 内置 bitmap 数据类型,原生支持高效率的 RoaringBitmap 各类操作函数,保证了计算的高效率。

通过实时计算Flink版 + Hologres 实时更新微批次的统计结果,同时基于 bitmap 丰富的表达能力和极致的压缩效率,实现了存储不膨胀,计算高效率和全实时。让亿级用户的 UV 计算,不再是门槛高、消耗资源的复杂操作。基于 bitmap 数据结构也可以实时还原原始的数据状态,确保了数据的高质量、可回溯。更灵活的数据结构,给了用户更灵活的分析体验,让自助式多维分析成为每个运营同学的必备工具。

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99元试用实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制T恤;另包3个月及以上还有85折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
相关文章
|
29天前
|
运维 监控 安全
选择主题1:实时计算Flink版最佳实践测评
本文介绍了使用实时计算Flink版进行用户行为分析的实践,涵盖用户行为趋势、留存分析、用户画像构建及异常检测等方面。与自建Flink集群相比,实时计算Flink版在稳定性、性能、开发运维和安全能力上表现更优,且显著降低了企业的IT支出和运维成本,提升了业务决策效率和系统可靠性,是企业级应用的理想选择。
76 32
|
30天前
|
运维 监控 安全
实时计算 Flink 版最佳实践测评
本文介绍了结合电商平台用户行为数据的实时计算Flink版实践,涵盖用户行为分析、标签画像构建、业务指标监控和数据分析预测等场景。文章还对比了实时计算Flink版与其他引擎及自建Flink集群在稳定性、性能、开发运维和安全能力方面的差异,分析了其成本与收益。最后,文章评估了实时计算Flink版的产品内引导、文档帮助、功能满足情况,并提出了针对不同业务场景的改进建议和与其他产品的联动可能性。
51 2
|
2月前
|
SQL 运维 监控
实时计算Flink版最佳实践测评报告
本报告旨在评估阿里云实时计算Flink版在实际应用中的表现,通过一系列的测试和分析来探讨其在稳定性、性能、开发运维及安全性方面的优势。同时,我们将结合具体的业务场景,如用户行为分析、标签画像构建等,来说明其实时数据处理能力,并对比自建Flink集群以及其他实时计算引擎。最后,从成本效益的角度出发,讨论采用全托管服务对企业运营的影响。
64 13
|
29天前
|
消息中间件 运维 分布式计算
实时计算Flink版最佳实践测评
本文介绍了使用阿里云实时计算Flink版进行用户行为分析的实践,详细探讨了其在性能、稳定性和成本方面的优势,以及与自建Flink集群的对比。通过实时计算,能够快速发现用户行为模式,优化产品功能,提升用户体验和市场竞争力。文章还提到了产品的易用性、功能满足度及改进建议,并与其他Flink实时计算产品进行了对比,强调了Flink在实时处理方面的优势。
|
2月前
|
存储 运维 监控
实时计算Flink版最佳实践测评
实时计算Flink版最佳实践测评
93 1
|
2月前
|
消息中间件 Java 数据库连接
Hologres 数据导入与导出的最佳实践
【9月更文第1天】Hologres 是一款高性能的实时数仓服务,旨在提供快速的数据分析能力。无论是从外部数据源导入数据还是将数据导出至其他系统,都需要确保过程既高效又可靠。本文将详细介绍如何有效地导入数据到 Hologres 中,以及如何从 Hologres 导出数据。
96 1
|
3月前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版产品使用问题之使用CTAS同步MySQL到Hologres时出现的时区差异,该如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
3月前
|
SQL 关系型数据库 MySQL
实时数仓 Hologres操作报错合集之Flink CTAS Source(Mysql) 表字段从可空改为非空的原因是什么
实时数仓Hologres是阿里云推出的一款高性能、实时分析的数据库服务,专为大数据分析和复杂查询场景设计。使用Hologres,企业能够打破传统数据仓库的延迟瓶颈,实现数据到决策的无缝衔接,加速业务创新和响应速度。以下是Hologres产品的一些典型使用场景合集。
|
4月前
|
存储 SQL OLAP
分析性能提升40%,阿里云Hologres流量场景最佳实践
分析性能提升40%,阿里云Hologres流量场景最佳实践
|
3月前
|
SQL 流计算
Flink SQL 在快手实践问题之使用Dynamic Cumulate Window绘制直播间累计UV曲线如何解决
Flink SQL 在快手实践问题之使用Dynamic Cumulate Window绘制直播间累计UV曲线如何解决
60 1

相关产品

  • 实时计算 Flink版