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、Office365、Skype等年收入270+亿美元的业务在监控数据质量方面的可靠技术保障。同时，基于业务需要，我们在设计和实现中达成下面一系列的目标：

● 监控流式数据的完整性与时延；

● 需要监控的数据管道(pipeline)具有多个数据生产者、多处理阶段、多数据消费者的特性；

● 数据质量的监控需要近实时(near real time)；

● 数据质量发生问题的时候，需要提供相应的诊断信息来帮助工程师迅速解决问题；

● 监控平台的服务本身需要超级稳定和高可用, 大于99.9%在线时间；

● 监控与审计本身是高度可信；

● 平台架构可以水平扩展 (Scale out)。

二、背景以及问题的引入

为了服务微软的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.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的审计元数据，一旦发生数据质量问题，工程师可以快速定位是哪个数据中心的哪台服务器在什么时间段发生了问题，然后快速采取相应行动来解决或缓解问题，并把对下游数据处理的影响降到最低。



可被监控的数据质量问题可以分为如下几类：

● 数据时延超出规
a660
定的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(M1*Mean, M2*Standard Deviation), M3*Mean) (M1