TOP100summit：【分享实录-Microsoft】基于Kafka与Spark的实时大数据质量监控平台

本篇文章内容来自2016年TOP100summit Microsoft资深产品经理邢国冬的案例分享。
编辑：Cynthia

邢国冬（Tony Xing）：Microsoft资深产品经理、负责微软应用与服务集团的大数据平台构建，数据产品与服务.

导读：微软的ASG (应用与服务集团）包含Bing,、Office,、Skype。天天产生多达5 PB以上数据，如何构建一个高扩展性的data audit服务来保证这样量级的数据完整性和实时性很是具备挑战性。本文将介绍微软ASG大数据团队如何利用Kafka、Spark以及Elasticsearch来解决这个问题。

一．案例简介

本案例介绍了微软大数据平台团队设计和部署的基于开源技术（Kafka、Spark、ElasticsSearch、Kibana）的大数据质量监控平台，这个平台具备实时、高可用、可扩展、高度可信的特性，成为微软Bing、Office36五、Skype等年收入270+亿美圆的业务在监控数据质量方面的可靠技术保障。同时，基于业务须要，咱们在设计和实现中达成下面一系列的目标：

● 监控流式数据的完整性与时延；
● 须要监控的数据管道(pipeline)具备多个数据生产者、多处理阶段、多数据消费者的特性；
● 数据质量的监控须要近实时(near real time)；
● 数据质量发生问题的时候，须要提供相应的诊断信息来帮助工程师迅速解决问题；
● 监控平台的服务自己须要超级稳定和高可用, 大于99.9%在线时间；
● 监控与审计自己是高度可信；
● 平台架构能够水平扩展 (Scale out)。

2、背景以及问题的引入

为了服务微软的Bing、Office 365以及Skype业务，咱们的大数据平台须要处理天天高达十几PB级别的海量大数据，全部的数据分析、报表、洞见以及A/B测试都依赖于高质量的数据，若是数据质量不高的话，依赖数据作决策的业务都会受到严重影响。

与此同时，微软业务对于实时数据处理的需求也日益增长，之前监控批处理数据（batch data）的不少解决方案已经再也不适用于实时的流式数据的质量监控。

在另一个层面，基于历史缘由，各个业务集团每每使用不一样的技术、工具来作数据处理，怎么整合这样异构的技术、工具以及在此之上的数据质量监控也是一个急需解决的问题。

图1是咱们数据处理平台的一个概念性架构。从数据生产者这端，咱们经过在客户端以及服务端使用通用的SDK，按照通用的schema来产生数据，数据经过分布在全世界的数据收集服务(collectors)来分发到相应的Kafka,而后经过pub/sub模式由各类各样的计算以及存储框架来订阅。

这样各类团队就能够选择他们最熟悉或者一直以来使用的工具来作处理。例如，从实时处理的角度，各个业务团队能够选用好比Spark或者微软的USQL streaming处理框架，以及其余第三方的工具来作一些特定场景的分析，好比日志分析的Splunk、交互式分析的Interana等。在批处理框架上，用户能够选用开源社区的Hadoop,、Spark或者微软的Cosmos等。

如图2所示，咱们在迁移大数据到图1架构的过程当中，也看到实时流式数据的快速增加。天天峰值消息高达一万亿个以上，每秒处理一百三十万个消息， 天天处理3.5PB流式数据。

3、数据监控的场景以及工做原理

3.1数据监控场景

基于业务需求，咱们总结归纳了须要被监控的数据处理管道特性（如图3）
● 多数据生产者（multiple data producers），数据来自客户端和服务端；
● 多个数据消费者(multiple data consumers)，这里特指各类数据处理框架；
● 多数据监控阶段（multiple stages），从数据产生到数据处理，数据每每流经多个数据管道的组件，咱们须要经过监控确保每一个阶段数据都不会发生丢失、高时延、以及异常。

3.2工做原理

基于图3的数据管道，咱们把问题具体化为如何确保基于Kafka的数据管道上下游的数据完整性、实时性、数据异常的监测。图4是一个抽象化的监控架构以及工做原理。

蓝色组件是数据管道里数据流经的各个处理阶段；绿色组件是本文中实时数据质量监控的核心服务Audit Trail。在数据流经各个组件的同时，相应的审计(audit)数据也会同时发到Audit Trail, 这个审计数据能够看做是一种元数据(meta data)，它包含关于数据流的信息，例如该消息是在哪一个数据中心、哪台机器产生；该消息包含几条记录、大小、时间戳等。Audit Trail汇总了各个数据处理组件发来的元数据后，就能够实时作各类数据质量的评估，好比数据在此时刻的完整性如何、实时性如何、有无异常。

基于图5的审计元数据，一旦发生数据质量问题，工程师能够快速定位是哪一个数据中心的哪台服务器在什么时间段发生了问题，而后快速采起相应行动来解决或缓解问题，并把对下游数据处理的影响降到最低。

可被监控的数据质量问题能够分为以下几类：
● 数据时延超出规定的SLA (service level agreement)

工程师能够经过如图6所示的时延状态图快速了解在数据质量时延这个维度是否正常，这对于对实时性要求比较严格的数据产品及应用很是重要，若是数据延迟到来，不少时候就失去了意义。

须要注意的是，图表在这里起到的只是辅助做用，在真正的生产环境中是经过系统API调用来按期检查SLA的符合状况，一旦超出时延阈值，会经过电话、短信等手段通知值班的工程师来实时解决问题。

● 数据在移动中发生丢失致使完整性不知足SLA (service level agreement)

工程师能够经过图7中所示简单图表来了解数据完整性的状态，图7所示包含两个数据处理阶段：一个数据生产者和两个数据消费者的应用案例。因此图表中其实是三条线，绿色是生产者的实时数据量，蓝色和紫色线是两个数据消费者处理的数据量。若是在理想状况下，数据完整性没有问题，这三条线是彻底重合。本例中在最后一个点出现了分叉，表明数据完整性出现问题，须要工程师进行干预。

● 数据自己发生异常-经过异常检测来实时监控

数据自己发生异常，咱们由相应的基于统计元数据的异常检测（如图8）来作实时监控。异常检测是一个在工业界很是广泛的问题和挑战，几乎每一个互联网公司都会有作异常检测的服务或平台，可是作好很不容易，这是一个能够单独写一篇文章的大题目，这里只是单辟一个章节作简单的算法介绍。

本例是经过对于数据量的异常检测来发现上游写log问题，或者其余数据生产的逻辑问题。

3.3异常检测

3.3.1异常检测算法1

咱们采用了Holt-Winters算法（图9）来训练模型和作预测，并在此之上作了不少改进来增长算法的强健性和容错能力。

强健性上的改进包括：
● 使用Median Absolute Deviation (MAD) 获得更好的估值；
● 处理数据丢点和噪声 (例如数据平滑)。
功能上的改进包括：
● 自动获取趋势和周期信息；
● 容许用户人工标记和反馈来更好的处理趋势变化。
经过比较预测值和实际值，咱们采用GLR (Generalized Likelihood Ratio) 来发现异常点。在这上面咱们也作了相应的改进，包括：
● Floating Threshold GLR, 基于新的输入数据动态调整模型；
● 对于噪声比较大的数据作去除异常点。

3.3.2异常检测算法2

这是一个基于Exchangeability Martingale的在线时间序列的异常检测算法，其核心就是假设数据的分布是稳定的。若是新的数据点的加入致使数据的分布(distribution)发生比较大的变化，咱们就认为异常发生了。因此基于历史数据，咱们须要定义一个新值异常公式（New value strangeness）。下面是这些公式的构成，对数学不感兴趣的读者能够略去。

在某个时刻t, 咱们收到一个新的数据点，对于历史每一个数据i：
s[i] = strangeness function of (value[i], history)
Let p[t] = (#{i: s[i] > s[t]}+ r*#{i: s[i]==s[t]})/N, where r is uniform in (0,1)
Uniform r makes sure p is uniform
Exchangeability Martingale: Mt=i=1tϵpiϵ-1
EMtp1,p2,…pt-1=Mt-1
Integrate ϵpiϵ-1 over [0,1] and pi is uniform
报警触发门槛经过Doob’s maximal inequality控制
Prob (∃ t :Mt>λ)<1λ
对于异常点，Martingale的值就会大于门槛值。

3.3.3异常检测算法3

这是一个简单而很是有效的基于历史数据的指数平滑算法。
它首先基于历史数据生成动态上下界：

Threshold (width) = min(max(M1Mean, M2Standard Deviation), M3*Mean) (M1<M3)
Alert: |Value – predicated value| > Threshold
预测值 = S1+12S2+14S3+18S4+116S51+12+14+18+116
优势在于处理周期性数据的异常检测很好，而且容许用户反馈和标记来调整动态上下界。

4、系统设计概述

基于业务场景的须要，咱们在设计和实现中须要达成一系列的目标以及处理相应的挑战：
● 监控流式数据的完整性与时延；
● 须要监控的数据管道(pipeline)具备多个数据生产者、多处理阶段、多数据消费者的特性；
● 数据质量的监控须要近实时(near real time)；
● 数据发生问题的时候，提供相应的诊断信息来帮助工程师迅速解决问题；
● 监控平台的服务自己须要超级稳定和高可用, 99.9%以上在线时间；
● 监控与审计自己是高度可信；
● 平台架构能够水平扩展 (Scale out)。

4.1高可用可扩展的架构

如图10所示，审计元数据经过前端服务（front end web service）到达Kafka, 咱们利用Kafka来实现高可用的临时存储（transient storage）, 这样，咱们的数据生产者和消费者在发送审计数据的同时，就不会发生阻塞进而影响更重要的数据流。

经过Spark streaming的应用，把审计数据按照时间窗口聚合，同时有相应的逻辑处理去重，晚到以及非顺序到来的数据，同时作各类容错处理保证高可用。

ElasticsSearch做为存储聚合的审计数据，经过Kibana作报表展现，进而经过Data Analysis service对外提供API来使得用户获取各类数据质量信息。

Data Analysis Service做为最终的API端，提供各类数据完整性、实时性、异常的信息。
上述组件，每一个都设计成能够独立水平扩展(Scale out), 而且在设计上保证高容错已实现高可用性。

4.2异地双活的可靠性保障

经过双数据中心Active-Active灾备（Disaster recovery）如图11所示，来进一步保证高可用高可靠的服务。总体架构保证数据流同时经过两个同构的审计处理管道进行处理，即便一个数据中心由于各类缘由下线，总体服务仍是处于可用状态，进而保证全天候的数据质量审计与监控。

4.3高度可信的审计与监控服务

对于任何监控服务来讲，常常被质疑的就是是否监控服务自己的结果是准确可信的。为了保证这一点，咱们经过两种方式来保证服务的可信度：
● 用来审计自身（Audit for audit）（图12）；
● Synthetic probe。

在基于Kafka/Spark/ES的管道以外，咱们还有一套独立的经由ES的审计元数据的处理管道，经过比较上述两个管道的结果，咱们就能保证审计数据的可靠性。
另外，基于synthetic probe的方式，咱们每分钟会发送一组synthetic数据进入前端服务（front end web service）, 而后试图从Data Analysis web service 读出，经过这种方式进一步保障数据的可靠性。

4.4辅助数据质量问题的诊断

当数据质量发生问题，Audit Trail提供了原始的审计元数据来帮助工程师进一步作问题的诊断。工程师可使用这些元数据和他们本身的trace来进一步JOIN, 来提供一种交互式的诊断，如图13。

5、效果评估和总结

经过上述系统架构的设计与部署，咱们实现了一系列支持公司Bing,、Office,、Skype业务发展的数据质量监控目标：
● 监控流式数据的完整性与时延；
● 须要监控的数据管道(pipeline)具备多个数据生产者、多处理阶段、多数据消费者的特性；
● 数据质量的监控须要近实时(near real time)；
● 数据发生问题的时候，须要提供相应的诊断信息来帮助工程师迅速解决问题；
● 监控平台的服务自己须要超级稳定和高可用, 99.9%在线时间
● 监控与审计自己是高度可信；
● 平台架构能够水平扩展 (Scale out)。

11月9-12日，北京国家会议中心，第六届TOP100全球软件案例研究峰会，Microsoft Principal Product Designer Bill Zhong将分享《微软OneNote的敏捷UX转型实践》；微软 data scientist Kirk Lee将分享《reinforcement learning in azure customer engagement》；微软 亚洲研究院资深研究员郑宇将分享《用大数据和AI驱动智能城市》。

TOP100全球软件案例研究峰会已举办六届，甄选全球软件研发优秀案例，每一年参会者达2000人次。包含产品、团队、架构、运维、大数据、人工智能等多个技术专场，现场学习谷歌、微软、腾讯、阿里、百度等一线互联网企业的最新研发实践。

大会开幕式单天体验票申请入口