通过flink实现个推海量消息数据的实时统计

关键词：

背景

消息报表主要用于统计消息任务的下发情况。比如，单条推送消息下发APP用户总量有多少，成功推送到手机的数量有多少，又有多少APP用户点击了弹窗通知并打开APP等。通过消息报表，我们可以很直观地看到消息推送的流转情况、消息下发到达成功率、用户对消息的点击情况等。

个推在提供消息推送服务时，为了更好地了解每天的推送情况，会从不同的维度进行数据统计，生成消息报表。个推每天下发的消息推送数巨大，可以达到数百亿级别，原本我们采用的离线统计系统已不能满足业务需求。随着业务能力的不断提升，我们选择了Flink作为数据处理引擎，以满足对海量消息推送数据的实时统计。

本文将主要阐述选择Flink的原因、Flink的重要特性以及优化后的实时计算方法。

离线计算平台架构

在消息报表系统的初期，我们采用的是离线计算的方式，主要采用spark作为计算引擎，原始数据存放在HDFS中，聚合数据存放在Solr、Hbase和Mysql中：

查询的时候，先根据筛选条件，查询的维度主要有三个：

1. appId
2. 下发时间
3. taskGroupName

根据不同维度可以查询到taskId的列表，然后根据task查询hbase获取相应的结果，获取下发、展示和点击相应的指标数据。在我们考虑将其改造为实时统计时，会存在着一系列的难点：

1. 原始数据体量巨大，每天数据量达到几百亿规模，需要支持高吞吐量；
2. 需要支持实时的查询；
3. 需要对多份数据进行关联；
4. 需要保证数据的完整性和数据的准确性。

Why Flink

Flink是什么

Flink 是一个针对流数据和批数据的分布式处理引擎。它主要是由 Java 代码实现。目前主要还是依靠开源社区的贡献而发展。

对 Flink 而言，其所要处理的主要场景就是流数据。Flink 的前身是柏林理工大学一个研究性项目， 在 2014 被 Apache 孵化器所接受，然后迅速地成为了 ASF（Apache Software Foundation）的顶级项目之一。

方案对比

为了实现个推消息报表的实时统计，我们之前考虑使用spark streaming作为我们的实时计算引擎，但是我们在考虑了spark streaming、storm和flink的一些差异点后，还是决定使用Flink作为计算引擎：

针对上面的业务痛点，Flink能够满足以下需要：

1. Flink以管道推送数据的方式，可以让Flink实现高吞吐量。

2. Flink是真正意义上的流式处理，延时更低，能够满足我们消息报表统计的实时性要求。

3. Flink可以依靠强大的窗口功能，实现数据的增量聚合；同时，可以在窗口内进行数据的join操作。

4. 我们的消息报表涉及到金额结算，因此对于不允许存在误差，Flink依赖自身的exact once机制，保证了我们数据不会重复消费和漏消费。

Flink的重要特性

下面我们来具体说说Flink中一些重要的特性，以及实现它的原理：

1）低延时、高吞吐

Flink速度之所以这么快，主要是在于它的流处理模型。

Flink 采用 Dataflow 模型，和 Lambda 模式不同。Dataflow 是纯粹的节点组成的一个图，图中的节点可以执行批计算，也可以是流计算，也可以是机器学习算法。流数据在节点之间流动，被节点上的处理函数实时 apply 处理，节点之间是用 netty 连接起来，两个 netty 之间 keepalive，网络 buffer 是自然反压的关键。

经过逻辑优化和物理优化，Dataflow 的逻辑关系和运行时的物理拓扑相差不大。这是纯粹的流式设计，时延和吞吐理论上是最优的。

简单来说，当一条数据被处理完成后，序列化到缓存中，然后立刻通过网络传输到下一个节点，由下一个节点继续处理。

2）Checkpoint

Flink是通过分布式快照来实现checkpoint，能够支持Exactly-Once语义。

分布式快照是基于Chandy和Lamport在1985年设计的一种算法，用于生成分布式系统当前状态的一致性快照，不会丢失信息且不会记录重复项。

Flink使用的是Chandy Lamport算法的一个变种，定期生成正在运行的流拓扑的状态快照，并将这些快照存储到持久存储中（例如：存储到HDFS或内存中文件系统）。检查点的存储频率是可配置的。

3）backpressure

back pressure出现的原因是为了应对短期数据尖峰。

旧版本Spark Streaming的back pressure通过限制最大消费速度实现，对于基于Receiver 形式，我们可以通过配置spark.streaming. receiver.maxRate参数来限制每个 receiver 每秒最大可以接收的记录的数据。

对于 Direct Approach 的数据接收，我们可以通过配置spark.streaming. kafka.maxRatePerPartition 参数来限制每次作业中每个 Kafka 分区最多读取的记录条数。

但这样是非常不方便的，在实际上线前，还需要对集群进行压测，来决定参数的大小。

Flink运行时的构造部件是operators以及streams。每一个operator消费一个中间/过渡状态的流，对它们进行转换，然后生产一个新的流。

描述这种机制最好的类比是：Flink使用有效的分布式阻塞队列来作为有界的缓冲区。如同Java里通用的阻塞队列跟处理线程进行连接一样，一旦队列达到容量上限，一个相对较慢的接受者将拖慢发送者。

消息报表的实时计算

优化之后，架构升级成如下：

可以看出，我们做了以下几点优化：

1. Flink替换了之前的spark，进行消息报表的实时计算；
2. ES替换了之前的Solr。

对于Flink进行实时计算，我们的关注点主要有以下4个方面：

1. ExactlyOnce保证了数据只会被消费一次
2. 状态管理的能力
3. 强大的时间窗口
4. 流批一体

为了实现我们实时统计报表的需求，主要依靠Flink的增量聚合功能。

首先，我们设置了Event Time作为时间窗口的类型，保证了只会计算当天的数据；同时，我们每隔一分钟增量统计当日的消息报表，因此分配1分钟的时间窗口。

然后我们使用.aggregate (AggregateFunction af, WindowFunction wf) 做增量的聚合操作，它能使用AggregateFunction提前聚合掉数据，减少 state 的存储压力。之后，我们将增量聚合后的数据写入到ES和Hbase中。

流程如下所示：

同时，在查询的时候，我们通过taskID、日期等维度进行查询，先从ES中获取taskID的集合，之后通过taskID查询hbase，得出统计结果。

总结

通过使用Flink，我们实现了对消息推送数据的实时统计，能够实时查看消息下发、展示、点击等数据指标，同时，借助FLink强大的状态管理功能，服务的稳定性也得到了一定的保障。未来，个推也将持续优化消息推送服务，并将Flink引入到其他的业务线中，以满足一些实时性要求高的业务场景需求。

个推数据统计产品（个数）ios集成实践

...数·应用统计”，根据官方的说法，个推的数据统计产品通过专业的移动应用数据分析，可以为用户的应用提供实时数据统计分析服务，包括了解版本质量、渠道状况、用户画像等。数据最后以可视化形式展现，很直观。我们尝... 查看详情

个推数据统计产品（个数）ios集成实践(代码片段)

...数·应用统计”，根据官方的说法，个推的数据统计产品通过专业的移动应用数据分析，可以为用户的应用提供实时数据统计分析服务，包括了解版本质量、渠道状况、用户画像等。数据最后以可视化形式展现，很直观。我们尝... 查看详情

个推techday直播回顾|分享基于flink的实时数仓搭建秘诀附课件下载

近日，个推TechDay“治数训练营”系列直播课第二期举办。来自每日互动（个推）的资深数据研发工程师为大家详细解读了实时数仓架构演进，分享了实时数仓的技术选型要点，并结合实战案例详细剖析实时数... 查看详情

个推techday直播回顾|分享基于flink的实时数仓搭建秘诀附课件下载

近日，个推TechDay“治数训练营”系列直播课第二期举办。来自每日互动（个推）的资深数据研发工程师为大家详细解读了实时数仓架构演进，分享了实时数仓的技术选型要点，并结合实战案例详细剖析实时数... 查看详情

flink实时计算pvuv的几种方法(代码片段)

...跑着不会停,所以要定期清理内存里的过时数据；收到的消息里的时间字段并不是按 查看详情

指标统计：基于流计算oceanus(flink)实现实时uvpv统计

...ff0c;腾讯CSIG高级工程师导语|最近梳理了一下如何用Flink来实现实时的UV、PV指标的统计，并和公司内微视部门的同事交流。然后针对该场景做了简化，并发现使用FlinkSQL来实现这些指标的统计会更加便捷。一、解决方案描... 查看详情

大数据flink实时大屏统计(代码片段)

目录1需求2数据3编码步骤:4代码实现5效果1需求Flink模拟双十一实时大屏统计在大数据的实时处理中，实时的大屏展示已经成了一个很重要的展示项，比如最有名的双十一大屏实时销售总价展示。除了这个，还有一些其... 查看详情

不惧流量持续上涨，bigo借助flink与pulsar打造实时消息系统

...性和稳定性提出了更高的要求。开源的Kafka集群难以支撑海量数据处理场景，我们需要投入更多的人力去维护多个K 查看详情

基于spark和flink的电商数据分析项目(代码片段)

...内的点击量趋势：各广告最近1小时内各分钟的点击量Flink实现业务需求用户访问session该模块主要是对用户访问session进行统计分析，包括session的聚合指标计算、按时间比例随机抽取session、获取每天点击、下单和购买排名前10的品... 查看详情

实时监控：基于流计算oceanus(flink)实现系统和应用级实时监控(代码片段)

...讯云大数据组件来完成实时监控系统的设计和实现，通过实时采集并分析云服务器（CVM）及其App应用的CPU和内存等资源消耗数据，以短信、电话、微信消息等方式实时反馈监控告警信息，高效地保障系统稳健... 查看详情

flink模拟项目：实时热门商品统计(代码片段)

首先要实现的是实时热门商品统计，我们将会基于UserBehavior数据集来进行分析。项目主体用Scala编写，采用IDEA作为开发环境进行项目编写，采用maven作为项目构建和管理工具。首先我们需要搭建项目框架。2.1创建Maven项目2.1.1项目... 查看详情

mongodb数据库批量插入海量数据时为啥有少部分数据丢失

...般这是不可能实现的，mongodb的数据存储在内存中，可以通过修改持久化参数SafeMode的safe改为true，即可直接持久化到文件系统中了。 查看详情

flink系列之：flinkcdc实现海量数据入湖

Flink系列之：FlinkCDC实现海量数据入湖一、历史数据入湖架构二、核心需求三、选择FlinkCDC原因四、FlinkCDC2.0原理五、支持全量与增量日志流并行读取一、历史数据入湖架构二、核心需求全量增量自动切换，并保证数据准确... 查看详情

如何实现消息推送功能

消息推送（Push）就是通过服务器把内容主动发送到客户端的过程。运营人员通过自己的产品或第三方工具对用户移动设备进行主动消息推送。完成推送后，消息通知会展示在移动设备的锁定屏幕及通知栏上，用户点击通知即可... 查看详情

如何使用消息推送功能？

...式来进行消息推送，比如使用个推消息推送服务。开发者通过集成个推消息推送SDK，即可简单、快捷地实现Android和iOS平台的消息推送功能，有效提高产品活跃度、增加用户留存。个推作为国内移动推送领域的早期进入者，于2010... 查看详情

flink系列之：基于scala语言实现flink实时消费kafkatopic中的数据(代码片段)

Flink系列之：基于scala语言实现flink实时消费KafkaTopic中的数据一、引入flink相关依赖二、properties保存连接kafka的配置三、构建flink实时消费环境四、添加Kafka源和处理数据五、完整代码六、执行程序查看消费到的数据一、引入fli... 查看详情

二.flink实时项目电商用户行为之实时流量统计(代码片段)

1.1模块创建和数据准备在Flink-project下新建一个mavenmodule作为子项目，命名为gmall-network-flow。在这个子模块中，我们同样并没有引入更多的依赖，所以也不需要改动pom文件。在src/main/目录下，将apache服务器的日志文件apache.log复制... 查看详情

flink+kafka实现wordcount实时计算(代码片段)

...流数据和批数据的分布式处理引擎。它主要是由Java代码实现。目前主要还是依靠开源社区的贡献而发展。对Flink而言，其所要处理的主要场景就是流数据，批数据只是流数据的一个极限特例而已。再换句话说，Flink会把所有任务... 查看详情