概述

背景

以互联网行业来说，在移动互联网发展比较成熟的现在，流量见顶，红利消失，企业竞争日趋惨烈，获取新增用户的成本日益增高。很多企业开始意识到不能一味的通过补贴、价格战、广告投放这种简单粗暴的方式抢占市场，这样的运作模式很难长时间维系。而通过精细化和数据化运营来降低成本、提升效率、最大化单用户价值的理念逐渐被越来越多的企业所接受。精细化和数据化运营的前提是要建立起一套完善的数据指标体系，借助这个数据指标体系企业可以有多方面的用途:

随着企业对数据化运营重视程度的日益增加，必然会衍生出大量的数据统计需求。而XL-LightHouse是以流式大数据统计为切入点，推动流式统计在诸多行业内的快速普及和大规模应用，定位是以一套服务使用较少的服务器资源同时支撑数以万计、数十万计的流式数据统计需求的大数据平台，致力于应对这种呈现"井喷"态势的流式数据统计需求所带来的一系列问题，寄希望于通过更加贴合场景、更具有实用价值的技术方案帮助企业降低数据化运营方面的成本。

收益

XL-LightHouse代表着一种以通用型流式大数据统计技术为切入点，低成本实现企业数据化运营的理念。它可以帮助企业更快速的搭建起一套较为完善的、稳定可靠的数据化运营体系，节省企业在数据化运营方面的投入，主要体现在以下几个方面：

架构

XL-LightHouse包括如下几个模块：

系统设计

XL-LightHouse是通用型流式大数据统计平台，它将流式数据统计需求抽象分类成多种运算场景，并对各种运算场景进行高性能的实现从而让每一种运算可以达到无限制复用的效果。

XL-LightHouse使用【统计工程-统计组-统计项】的三层结构来管理所有统计需求。每一个统计需求叫做一个统计项，每个统计项都是基于一种或多种运算场景。用户可根据需要创建若干个统计工程，每个统计工程可包含多个统计项，而基于同一份元数据的多个统计项叫做一个统计组。

Web模块可管理统计项的运行状态，用户可在Web端页面启动、停止、删除指定的统计项，处于运行状态的统计项正常执行统计运算，非运行状态的统计项不执行统计运算。接入系统首先需要用户在Web端进行相应配置，然后通过SDK上报原始数据。系统将统计原始消息数据按照统计周期划分成若干个批次再依据统计配置进行相应计算。

1、自定义流式统计规范(XL-Formula)

SQL规范在大数据查询和统计分析方面被广泛应用，SQL在离线数据分析、OLAP、OLTP等领域都具有不可撼动的地位。而且随着FlinkSQL和SparkSQL等组件功能的日趋完善，SQL在流式统计领域也开始被越来越多的使用。但是由于SQL本身是基于数据表的概念进行数据处理，不可避免需要存储较多的原始数据和中间态数据在内存中，造成较高的内存浪费；分布式SQL在数据处理过程中会触发Shuffle，造成大量的网络传输，影响执行效率；SQL在一些分组聚合操作可能引起较为严重的数据倾斜，对程序的正常执行造成影响，很多SQL计算任务需要依据数据量和运算逻辑进行特定优化；针对特定的统计需求需要执行单独的计算任务，不同统计任务之间运算资源不能共用，从而造成较高的计算资源浪费；SQL语法过于臃肿和复杂、不够清晰简洁、多过滤条件的组合逻辑需要依赖较长的SQL语句来实现，不便于理解，书写较长SQL语句容易出错；SQL函数定制化功能扩展不够方便；SQL开发相对较复杂，实现相同功能SQL可能会有多种写法，不同写法执行和解析效率也各有差异。这些问题使得相应功能的实现需要依赖专业的数据研发人员，导致流式统计任务研发成本高、周期长。当企业数据指标呈现指数级增长时，SQL规范的瓶颈也将凸显出来，需要耗费大量的研发成本、数据维护成本和服务器运算成本。我认为SQL规范的这些问题限制了它在流式统计这个细分场景内的快速扩张，使得SQL在这个细分领域内的应用基本局限在定制化需求开发的范围之内。从一定程度上来说SQL规范已经阻碍了流式统计的发展，制约了流式统计在各行业内的快速普及和大规模应用。XL-LightHouse作为一个通用型流式大数据统计平台，侧重于帮助企业解决繁杂的流式数据统计问题。XL-LightHouse并没有拘泥于现行的大数据领域的业内标准，而是寄希望通过使用更为轻巧的技术方案解决目前企业所面对的问题。它定义了一套较为完善的用于描述形式各样的流式统计需求的配置规范，通过各个属性的组合可以实现非常强大的统计功能，从而帮助企业更快速的搭建起一套较为完善的、稳定可靠的数据化运营体系。

2、消息聚合处理

系统将整个数据消费链路分成以下基本环节：Client模块上报消息数据环节、RPC模块处理消息数据环节、运算模块执行展开和分组操作环节、统计结果存储环节。在每个环节系统使用异步处理、批量消费、对重复性计算进行聚合处理的方案。各环节接收到消息后放入消息缓冲池，系统依据各环节的预定义聚合逻辑将消息划分成不同的计算类型，对单节点单进程内相同类型的消息进行聚合处理。这种设计可以减少数据向下游传输、提升网络IO效率、又可以直接减少下游运算量以及DB的写入压力。从Client端发送消息到最终的统计结果入库中间的每个环节都对重复性消息进行聚合处理尽可能减少消息量，并且将与下游运算无关的参数都会尽早抛弃掉，XL-LightHouse的数据消费链路是一个逐层递减的结构。各个环节的消息聚合逻辑略有不同，以Client模块为例消息聚合主要包括以下内容：

（1）消息体参数裁剪

为了提高消息的传输速度并提升后续步骤消息聚合效率，Client模块需要对原始消息进行裁剪操作，其目的是去掉统计无关字段。统计无关字段是系统根据各统计组下所有有效统计项计算得来，对于与所有有效统计项均不相关的字段在Client模块上报数据之前将其过滤掉，避免非必要的数据传输。

（2）篡改消息体时间戳

Client模块上报消息环节在执行聚合操作前修改消息原始时间戳为最小批次时间，其目的是为了后续步骤中在保证数据准确性的前提下能够将尽可能多的消息聚合到一起，减少网络传输和下游运算量。Client模块以当前统计组下所有有效统计项的统计周期的最大公约数为时间窗口，按照该时间窗口和消息原始时间戳计算得到消息所对应的最小批次时间。Client模块将消息原来的时间戳修改为最小批次时间然后放入缓冲池。

（3）聚合操作

聚合操作即为将同类型消息按预定义聚合逻辑合并到一起。不同环节的聚合逻辑略有不同，Client模块的聚合逻辑是指消息内容一致的消息，即为相同统计组、相同参数值的消息。原始消息发送到缓冲池后消费线程组定时从缓冲池中批量读取消息，并将其中符合聚合规则的消息聚合到一起。经过聚合操作后消息体的数据结构由单条消息体内容变更为消息体内容和消息体重复次数两个属性。

3、消息展开与分组

在XL-LightHouse中集群内的所有统计任务共用集群运算资源，运算模块接收到数据后对统计消息进行展开和分组操作。

在大多数业务场景中针对一份元数据往往有多个数据指标，统计组下的所有统计项共用一份原始数据消息。展开操作即为查询统计组下所有有效统计项，提取各统计项的关联字段，为各统计项复制一份单独的消息数据并只保留其运算相关字段的过程。展开操作的目的是为了避免各统计项的后续运算逻辑相互之间产生影响。

分组操作即为提取统计项的统计周期属性，依据统计周期划分时间窗口并按时间窗口对展开操作后的消息进行分组；然后判断统计项是否包含多个统计运算单元，如果包含多个统计运算单元则按统计运算单元进行再分组；判断统计项是否包含维度属性，如包含维度属性则提取维度信息并按维度进行再分组。分组操作的目的在于将各统计任务的运算过程进行分解，拆分成不同的计算类型，同类型消息聚合处理，不同类型的消息运算过程互不影响。

4、消息缓冲池

系统聚合处理所依赖的消息缓冲池实现方案基于有界优先阻塞队列。系统将消息缓冲池分成若干个Slot，每个Slot的组成结构包括一个BoundedPriorityBlockingQueue（有界优先阻塞队列）和Slot对应的最后访问时间戳。消息缓冲池的处理逻辑包括以下步骤：

（1）Producer按照不同环节的聚合逻辑生成消息事件的Key，Key用于区分是否为相同类型的消息；

（2）消息缓冲池依据消息Key按照Hash取余分配对应的Slot；

（3）按照预定义时间窗口将消息划分到不同的处理周期；

（4）Slot对相同处理周期的消息按照Key进行优先排序，不同处理周期的消息按窗口时间排序；

（5）消费线程组定时轮询各个Slot；

（6）判断Slot的使用容量是否超出阈值，阈值为batchsize * backlog_factor，其中batchsize为指定的单次消费最大消息数量，backlog_factor为指定的消息积压系数；

（7）如果Slot使用容量没有超出阈值，则继续判断Slot的上次消费访问时间，如果超出时间阈值则读取消息批量消费，否则跳过本次任务。 消费Slot消息后同时更新Slot使用容量以及最后访问时间。

该消息缓冲池实现可以将尽可能多的相同计算类型的消息聚合到一起处理，减少对下游运算量和DB的写入压力。

5、基数运算

bitcount基数运算是指distinct（非重复值数量统计），系统使用基数过滤装置过滤已存在的基数值，通过判定在过滤装置中不存在的基数数量然后更新DB中的统计结果从而实现基数统计。基数过滤装置包括内存基数过滤装置和分布式基数过滤装置两部分。内存基数过滤装置的作用在于初步判断基数值是否已存在，其作用在于内存判断效率更高，从而尽可能避免重复性的基数判断对整体性能的影响。内存基数过滤装置使用RoaringBitMap工具包实现。分布式基数过滤装置内含多个分片，每个分片对应一个RoaringBitMap数据存储结构，分片数可根据实际需要指定，通过提高分片数可以提高基数运算的准确度。分布式基数过滤装置的实现方案包括如下步骤：

（1）将原始数值经过MurmurHash-128Bit生成原始数值对应的Long类型的Hash值。

（2）设置统计任务所需的分片数，每个分片对应一个RoaringBitMap数据结构，本系统过滤装置采用Redis扩展Redis-Roaring插件的方式实现，原始数值对应的分片可通过Hash取余获得。

（3）将Long类型的Hash值按高32bit和低32bit拆分成两个Int类型整数，如果为负数取其绝对值，两个Int值的组合对应原始值在RoaringBitMap数据结构中的Index值。

（4）批量将多个基数值对应的Int值组合发送到Redis，将基数判断的多个操作使用Lua脚本合并执行。判断Int值组合是否在过滤装置中存在，如果两个Int值都在过滤装置中存在，则表示原始值已存在，否则为原始值不存在，如果原始值在过滤装置中不存在系统在判定完成后更新相应Index值。

（5）统计在过滤装置中不存在的原始值的数量并更新到DB中。

该实现方案的好处在于基数运算不需要存储原始值可减少对内存的占用；使用MurmurHash-128Bit生成Index值从而不需要维护原始数值和Index的映射关系；RoaringBitMap算法本身具有压缩位图功能可以减少基数稀疏情况下的内存浪费的问题；使用Lua脚本实现基数过滤功能可以减少对Redis的访问次数提升整体性能；使用内存基数过滤装置进行初筛可以避免不必要的重复判定；通过调整分片数可以很方便的提升基数统计的准确率。

6、避免shuffle

在大数据任务的执行过程中shuffle是影响性能比较主要的一个因素，Shuffle除了会带来大量的网络开销还可能引起数据倾斜甚至OOM等问题。系统采用避免Shuffle这种不可控的因素从而规避Shuffle可能带来的不可预料的问题。运算模块基于Structured Streaming开发，采用完全规避Shuffle的计算方式，通过设置运算节点数量调整任务执行并行度，系统将单运算节点内的统计消息依据统计项标识、维度标识、时间批次、统计运算单元拆分成不同的计算类型。统计结果数据和中间态数据基于外部存储实现。本系统中统计结果存储在Hbase中，bitcount基数运算的中间态数据存储在Redis中、limit运算的排序数据存储在Redis中。每个运算节点在运算过程中只与外部存储通信，不同运算节点之间互不影响。

7、统计限流

为了避免因为某个大数据量的统计需求的突然接入或某个统计项的流量暴涨而导致系统的不稳定，系统针对统计组消息量、统计项结果量、统计项运算量等维度的熔断保护机制。该限流保护机制的作用在于可以更好的保障整体服务的稳定性，目前包含以下策略：

（1）统计组消息量限流

统计组消息量限流是针对单位时间内接收到的统计组消息数量的限流策略。系统内置统计组消息量计数装置用于计算单位时间内接收到的统计组消息数量。当单位时间内消息量超出阈值后触发限流，使当前统计组进入限流状态。Client模块以及Tasks模块自动抛弃非正常状态下的统计组消息。由于一个统计组可对应一个或多个统计项，所以该限流策略会影响统计组下所有统计项的正常统计。统计组进入限流状态后在指定时间内（默认20分钟）自动抛弃相应消息，当限流时间达到时间阈值后统计组自动恢复到正常状态。

（2）统计项结果量限流

统计项结果量限流是针对单位时间内统计项生成的统计结果数量的限流策略。系统内置统计项结果量计数装置用于计算单位时间内生成统计结果的数量。当单位时间内结果量超出阈值后触发限流，使当前统计项进入限流状态。统计项结果量跟两个因素有关，一是统计周期的时间粒度，统计周期粒度越细的指标数据量越多，比如秒级和分钟级统计单位时间内生成的统计结果要多于小时级和天级的统计。第二个影响因素是维度，维度数量越多的统计项单位时间内生成的统计结果更多，比如以城市为维度的统计指标生成的统计结果量要高于以省份为维度的统计指标。统计项结果量限流是针对当前统计项的限流策略，所以只对当前统计项有影响，对统计组下其他统计项没有影响。当统计项进入限流状态后在指定时间内（默认20分钟）自动抛弃相应相应消息，当限流时间达到时间阈值后当前统计项自动恢复到正常状态。

8、时间戳压缩

系统针对流式统计场景对数据存储格式进一步优化，目的在于提高DB的数据吞吐量。系统统计结果数据存储采用时间戳压缩，根据统计周期划分成不同的时段，将每个统计项相同维度下的同一时段内的多个统计结果数值存储在不同的column内，列名采用delta压缩，同一时段内的数据使用相同的Key，减少Key值的重复。

9、异常熔断

熔断机制是为了保障业务方自身服务的稳定性，避免因统计服务的不稳定而对业务方自身服务产生影响。异常熔断机制是指在调用client接口时，如果单位时间内的失败次数或超时次数超出阈值，则进入熔断状态，此时client模块自动跳过统计消息发送逻辑。进入熔断状态后，client模块周期性检测统计服务状态是否恢复正常，如果统计服务恢复正常则自动重连。

系统功能边界

Hello World 使用范例

完整版使用示例请查阅：

范例一：首页ICON区域用户行为数据统计

<img src=https://cloud.tencent.com/developer/article/"https://lighthousedp-1300542249.cos.ap-nanjing.myqcloud.com/4301-2/1.png" width="300px" height="200px" />

该区域包含3个Tab，每个Tab有多个业务ICON图标，用户手动滑动可切换Tab，假设针对该ICON区域我们有如下数据指标需求：

用户点击ICON图标时上报相应埋点数据

范例二：移动支付订单数据统计

1、 支付成功订单数据统计

用户支付成功后上报原始消息数据。

2、 订单支付状态数据监控

我这里假设订单有四种状态：支付成功、支付失败、超时未支付、订单取消。

更多适用场景举例

项目地址

交流反馈

一键部署

写在最后的一些话

XL-LightHouse是一套通用型流式大数据统计平台，致力于推动流式统计技术的快速普及和大规模应用，定位是以一套服务使用较少的服务器资源同时支撑数以万计、数十万计流式数据统计需求的大数据平台。XL-LightHouse面向企业自上而下所有职能人员共同使用，倡导以通用型流式数据统计为切入点，倾向于选择更为轻巧的技术方案帮助企业更快速的搭建起一套犹如我们人体神经系统一样遍布全身的、较为完善稳定可靠的数据化运营体系。

流式统计技术并不完美，确实有一些场景不适合使用流式统计实现，所以它也不可能完全取代了其他的技术方案。但是我依然认为在企业数据化运营领域在所有的技术方案中，能够发挥中流砥柱作用的只有可能是通用型流式数据统计。时效性是流式统计得以青睐的一个原因，但我认为最根本原因在于一项技术能够普及到什么程度，很多时候使用的成本决定了一切。

在软件研发领域，我认为通用型流式统计将会对现在的软件类产品研发产生较为巨大的影响，它会发展成为如同日志一样的重要角色，通用型流式统计或将成为独立于日志之外且重要程度不亚于日志的另一套辅助类工具体系，各种工种的程序员将会在任何有必要的地方加上流式统计的代码就像加日志一样司空见惯、习以为常。

在企业级服务市场，我相信通用型流式数据统计将凭借其庞大的应用场景和巨大的业务价值而成为企业最核心的基础服务之一，而以通用型流式数据统计为核心理念、以其他技术方案为辅助手段的数据化运营类产品将成为企业级B端市场不可或缺的中坚力量。此外，伴随着软硬件技术的协同发展以及物联网时代的即将到来，我认为通用型流式数据统计也将渗透于现实世界各个方面，成为社会的一种基础运算能力，在各类行业中得到较为普遍的应用。