如果数据量大，维度多，用keyBy并不方便，建议写到外部实时数仓里，Clickhouse擅长实时查询，flink擅长实时处理。

一、多维度复杂统计（使用Clickhouse）

使用是clickhouse的ReplacingMergeTree，可以将同一个分区中，ID相同的数据进行merge，可以保留最新的数据，可以使用这个特点实现Flink + Clickhouse（勉强）实现数据一致性。

存在的问题：写入到clickhouse中的数据不能立即merge，需要手动optimize或后台自动合并。

解决方案：查询时在表名的后面加上final关键字，就只查最新的数据数据，但是效率变低了。

如何设计clickhouse的表？

1.可以支持维度查询（大宽表） 2.按照时间段进行查询（将时间作为表的字段并且建分区表） 3.可以统计出PV、UV（去重查询） 4.支持分区（按照时间进行分区） 5.支持覆盖（ReplacingMergeTree）（对查询结果准确性要求高的，表名后面加final） 6.如果生成一个唯一的ID （在Kafka中生成唯一的ID，topic+分区+偏移量） 7.相同的数据要进入到相同的分区（按照数据的时间即EventTime进行分区）

1、建表

2、自定义Kafka反序列器生成唯一ID

MyKafkaStringDeserializationSchema

FlinkUtils新增createKafkaStreamV2

调用TestKafkaId

3、使用JdbcSink将数据写入clickhouse

导入clickhouse驱动

JsonToBeanFuncV2

二、观看直播人数统计需求

a)实时统计累计观众（实时展示，直接用flink keyBy统计） b)实时统计各个直播间在线人数（实时展示，直接用flink keyBy统计） c)查看多个维度的明细（将数据写入到clickhouse中）

1、观看直播人数统计实现

实现方式一： a)将数据来一条就写入到Redis/MySQL或大屏展示（延迟低、但是效率低、对数据库压力大） b)再写一个job将各种明细数据写入到ClickHouse中（提交了2个job、数据重复计算）

AnchorDistinctTotalAudienceFunc

实现方式二： 将数据攒起来批量（不是简单的增量聚合，不能使用窗口，而是使用定时器）写入到Redis/MySQL（延迟高、效率高、对数据库的压力小） 在同一个job中，将数据写入到Clickhouse中（同一个主题（类型）的数据尽量在一个job中完成，将不同的数据打上不同的标签，侧流输出，可以节省集群资源。避免数据重复读取和计算）

AnchorDistinctTotalAudienceFunc

2、直播人气值计算

• 在直播间中至少停留1分钟
• 在30分钟之内，同一设备ID频繁进入该直播间，算一个用户的人气值

实现思路： 按照EventTime划分滑动窗口 使用processFunction注册定时器

3、观看直播人数统计结果保存

建表：

写入代码在上面的方式二

二、打赏礼物需求分析

在MySQL中还有一种礼物表（维表），需要进行关联，关联维表通常的解决方案：

a) 每来一条数据查一次数据库（慢、吞吐量低） b) 可以使用异步IO（相对快，消耗资源多） c) 广播State（最快、适用于少量数据、数据可以变化的）

礼物表：小表、数据可能会发生变化（礼物停用、新增了一个礼物）

统计的具体指标： 1.各个主播的收到礼物的数量、积分值 2.打赏礼物的数量、受欢迎礼物topN（可以将数据写入到数据库中，查询时再排序） 3.做多维度的指标统计（ClickHouse）

1、按照主播（直播间）统计礼物的积分（抖币）

统计的具体指标：各个主播的收到礼物的数量、积分值

礼物表使用jdbcSource, 不停的读数据 行为数据使用kafkaSource

MySQLSource

GiftConnectFunction

2、热门商品TopN需求分析

统计具体指标：统计10分钟内，每隔1分钟统计一次各个分类、各种事件类型的热门商品（商品ID）

HotGoodsTopNFunction

HotGoodsAggregateFunction

HotGoodsWindowFunction

ItemEventCount