第六章 大数据，6.2 双11背后的大规模数据处理(做者：惠岸 朋春 谦乐)

6.2 双11背后的大规模数据处理

1. 实时数据总线服务-TT

TimeTunnel（TT）在阿里巴巴集团内部是一个有着超过6年历史的实时数据总线服务，它是前台在线业务和后端异步数据处理之间的桥梁。从宏观方面来看，开源界很是著名的Kafka+Flume的组合在必定程度上可以提供和TT相似的基础功能；不一样的是，在阿里巴巴的业务体量和诉求下，咱们有比较多的配置管控、资源调度、轨迹校验和血缘识别等方面的工做。

TimeTunnel产品架构

 

1.1 Pub/Sub服务

经过上图咱们清楚地看到，TT的核心部分是一个基于HBase作中间存储的Pub/Sub服务，它提供了一个能支撑高读写比、大吞吐量和数据不丢的队列服务。除此以外，基于平常运维考虑，咱们还支持了按时间seek和弹性伸缩的能力。

数据须要在Pub/Sub“落地”的需求一方面来自于业务上对热点数据多份消费的考虑，另外一方面一些在线算法方面的应用须要常常性地对数据进行回放训练，数据“落地”可以比较好地对先后台进行解耦。事实上，TT里最热门的数据（例如天猫交易相关）有超过100倍的读写比；而从总体来看，仅双11当天流出TT的数据也比流入的数据多了3倍以上。

选择HBase做为中间存储的缘由是可以成本较低地复用基于HDFS的多副本存储能力，以及HBase自身在提供读写服务时对于热点数据的内存管理能力。图 8是写入TT的数据在HBase中的存储模型，咱们在broker层面经过构造合理的rowkey来使得同一个分区下的数据可按rowkey顺序scan；同时，由于在生成rowkey的时候咱们使用了broker上的时间戳做为高位变量，所以能很方便地提供按时间seek的能力。

数据在HBase中的存储模型

 

1.2数据采集

上图左侧黄色部分是TT的数据采集方案。咱们经过如下途径来准实时地收集前台业务产生的增量数据：

1. 依赖DRC实现对MySQL、OceanBase以及Oracle等前台业务数据库的增量变动进行捕捉解析；
2. 自研的日志Agent部署在数十万台的应用服务器上，准实时地捕捉应用日志的变化；
3. 和其余一些内部主流存储例如OTS进行打通；
4. 用户采用TT提供的SDK主动写入。

随着集团内重要业务异地多活架构和全球化的发展，数据采集分散在跨越数千甚至上万千米的多个IDC中；而与此相反，以Galaxy、ODPS为表明的大数据计算服务则须要考虑充分地利用大集中的架构来提高吞吐能力。所以，不可避免地在数据采集过程当中须要对数据进行缓冲和压缩以尽量下降长途链路对于吞吐量的负面影响。

矛盾的是，缓冲意味着前端产生的数据须要在采集端“等待”，也就意味着消费方看到的数据是延迟的。这对于像阿里妈妈这样依赖TT作反做弊和实时计费的业务来说是很难接受的，数据延迟意味着资损，意味着用户体验的显著降低。一样地，压缩也是要消耗采集端的服务器资源的，尤为在双11这样的场景下，前台业务对于采集端的功耗尤为敏感。

遗憾的是，世界上历来没有一个只带来好处而没有任何弊端的事物，软件和产品的设计中到处都是折衷和取舍。除了在技术层面将实现细节作到尽量极致，TT为了服务这些不一样的场景，也提供了一些可配置的参数例如buffersize、sendthreads、compressLevel等用来匹配用户对延时、性能以及功耗的不一样需求。

 

1.3 轨迹校验

TT区别于其余相似产品的最大之处，是咱们经过技术埋点实现了一套完整的数据轨迹校验的方案——咱们称之为“门将”。轨迹校验的目的在于经过监控的手段来保证“数据不丢”，设计得好，甚至能够识别出数据的重复、乱序等状况。

几乎全部相似的产品都宣称本身能作到“数据不丢”，固然也包括配备了“门将”以前的TT。有意思的是，几乎全部相似的产品都会被“丢数据”这个问题困扰，一样包括TT。由于我相信咱们必定有能力在软件设计以及编码实现方面作到“数据不丢”的承诺，但每每会在一些预期外的异常case、版本升级或者系统耦合的地方出现这样那样的纰漏，从而致使下游消费方看起来缺失了部分数据。

以日志采集为例，咱们碰到过由于操做系统的限制（请参阅max_user_watches相关的说明），inotify没有通知到新文件的产生而发生整个文件漏采集；也碰到过由于软件的bug在递归建立子目录的状况下出现了时序问题致使文件漏采集；还碰到过保存在应用服务器上的checkpoint文件被意外损坏致使的“丢数据”。这样的案例实在太多，并且防不胜防。

因此，工业界真正的“数据不丢”我认为是有完备的机制可以快速地发现数据丢失，考验的是系统的监控能力。

上文提到过，TT支撑着阿里妈妈的实时反做弊和点击计费业务；一样地，蚂蚁金服大量涉及资金核对和商户对帐的业务也将身家性命托付在TT上。这样的业务不容许有任何缘由致使的数据正确性问题。

“门将”的核心思路是在采集端往TT写入数据的同时，构造恰当的meta，将数据“链表化”，从而可以在“门将”的校验服务里对数据轨迹进行还原，进而和源头进行校验（图 8）。

仍然以日志采集为例。在采集过程当中，咱们以ip+dev+inode+sign来惟一识别内网上的一个文件，在构造meta时记录下当前数据包在原始文件中的offset和当前数据包的大小size，那么对于同一个文件的多个数据包，经过offset和size就能快速地识别出文件内有没有被重复采集或者遗漏采集。若是在恰当的时间内与这台机器上ls命令获得的结果进行比对，就很容易发现有没有文件被漏采集。

 

1.4 小结

全部的技术实现都是业务需求的抽象，这些需求有可能来自于大多数用户须要用到的功能，更有可能来自对上下游业务架构和场景的理解。数据总线服务是一个和业务架构耦合很是密切的基础组件，阿里巴巴集团独特的技术架构、多样性的存储方案和横向平台化的研发模式赋予了TT探究更复杂问题的原动力。

在2016年双11这样一个万众瞩目的时间点，TT经过前期的软件性能和机房规划上的努力，高峰期单一集群承担了15GB/s的写入和50GB/s的读取流量，首次作到了对全部业务进行不降级服务。这对于咱们、对于搭建在TT上的众多业务，都是极大的鼓舞。

 

2. Galaxy：大规模数据流处理技术

每一年双11除了“折扣”，阿里人关注的另外一个焦点，就是面向全世界媒体直播的“实时大屏”（以下图所示）。包括总成交量在内的各项指标，经过数字维度展示了双11狂欢节这一是买家，卖家及物流小二共同创造的奇迹！

图：双11媒体直播大屏

为实现这一大屏，背后须要实时处理海量的、庞大电商系统各个模块产生的交易日志。例如双11当天产生的日志量达到了PB级别，而每秒处理的峰值更是高达近1亿事件！

如此大规模、高吞吐和低延时计算，带来一系列世界级的技术挑战，包括：

1. 实时编程：流式的数据处理给业务逻辑的表达和推理带来了不少的复杂性。特别面对不断变化的业务需求，如何帮助用户快速地编写和验证明时计算逻辑是相当重要的。
2. 低延时：实时计算强调计算延时和结果的时效性。例如实时大屏对计算延时特别敏感，每一年的双11都超越前一年更早地达到相同的成交量，系统须要在秒级甚至毫秒级反应出每一笔交易。即便在流量高峰时（双11晚0:00点）也须要保证延时！
3. 集群利用率：为提升资源利用率，咱们将不用业务的实时处理逻辑共享一个集群。这样的共享也带来性能隔离的问题，即如何让同一台物理机上的不一样逻辑任务不互相干扰。这也是大部分开源框架忽略的重要问题。
4. 严格容错及数据一致性：随着应对高吞吐而不断扩大的集群规模，各类软硬件故障都难以免。如何保证明时计算在任何故障下都能产生准确、一致的计算结果，不遗漏、重复事件输出，也不引发内部状态的误差，是另外一个重大挑战。
5. 多样化场景支持：随着实时决策对业务的价值愈来愈多，系统还须要支持愈来愈复杂和多样化的场景，如在线机器学习、结合图计算实现的动态关系网络分析等等。

下文介绍Galaxy的重要技术创新，简要描述它们如何帮助应对以上技术挑战。

 

2.1 SQL与增量计算——复用熟悉的离线思惟，自动实现增量（流式）计算

为了简化用户编程，特别是利用原有的离线计算做业快速实现实时计算，Galaxy容许经过高层描述性语言，如用户熟悉的SQL来编写流计算做业。经过简单几行SQL代码就能够实现过滤、双流关联等业务逻辑。

在执行时，因为数据是以流式进入系统的，用户的SQL就像数据库视图同样，被自动增量更新，并以必定的频率输出结果，供下游计算和展现。

这一独特的编程设计，不只帮助用户借助熟悉的离线处理思惟表达实时计算逻辑，也由于一样的程序能够在离线系统运行，使得结果的对比变得易如反掌。

 

2.2 高性能优化引擎——实现低延时计算

用户的SQL脚本通过编译优化，生成数据流图，而后运行于Galaxy的分布式引擎之上。相比开源数据流引擎，Galaxy引擎在“阿里巴巴规模”下，面对真实复杂的业务场景作了不少优化。包括自适应的消息打包、自定义序列化、数据行+列压缩、先进的内存管理、和内部缓存队列和线程模型，以及基于下游向上游“反向”传递压力的流控策略等。

图：Galaxy优化执行流和运行时模块

通过以上一系列的优化，Galaxy相比去年提高了6倍左右的吞吐性能。下图显示了Galaxy相比开源系统的性能优点。在面对今年双11 3倍于去年的峰值状况下，表现很是稳健。

图：开源框架性能对比，经过“窗口WordCount（6组参数）”基准测试获取

 

2.3 灵活的资源调度

Galaxy面对阿里巴巴集团众多业务场景，将不一样业务放置于大规模（几千台服务器组成的）共享集群中，以提升资源利用率。另外一方面也随之带来了“多租户”环境下的做业资源隔离问题，它直接影响资源的有效利用和做业的计算性能。

通过多年的积累，Galaxy支持CPU、内存、网络和磁盘I/O等多维度资源的隔离。例如，对于CPU的隔离支持灵活的min-max策略，既保证了每一个做业最基本的资源需求，也使的空闲的资源被最大限度利用。

图：做业维度的CPU资源min-max共享模型

在此基础上，Galaxy的资源调度还支持必定比例的“超卖”、做业优先级调度、动态负载均衡和微做业共享单一物理核等多种机制。对于资源消耗特别大的做业还支持动态按需分配（即资源的弹性分配）。在知足复杂的运维要求和实时计算连续性的同时，实现了高效的资源利用和性能隔离。

 

2.4 容错与状态管理

流计算须要连续处理可能无界的输入和连续产生输出。在长时间运行中，大规模计算集群的各类软件或硬件故障难以免。由此对于计算和中间结果（如内存状态）的容错就相当重要。为了作到精确的容错和故障恢复，保证结果的准确性。Galaxy支持多种灵活的容错策略，以在不一样计算特性下，权衡容错资源消耗和恢复性能。如基于输入的从新计算、状态检查点（checkpoint），甚至是多副本的状态和计算容错等。

特别是自动的分布式增量检查点功能，系统自动利用内存、本地磁盘和远程存储构成的多级存储，在不影响流计算延时的状况下异步实现了计算状态的持久化。当有故障发生时，保存的状态能够被快速加载。这一切对用户都是无感知的。

图：自动利用多级存储的流计算状态管理

 

2.5 开放可编程API（兼容Apache Beam）

除了SQL这样高层的描述语言和用户自定义逻辑（UDF），Galaxy还支持Apache Beam API，以提供更为灵活的实时逻辑编程。Beam是一个统一开放的大数据应用编程接口，能够同时描述离线和在线逻辑，最先由Google提出。Beam提供了功能丰富的编程接口，能有效的处理有界、无界、乱序的数据流输入。 下面显示了经过Beam实现的流式WordCount的例子：

1.指定Runner(底层计算引擎)建立一个Pipeline。

2.使用Source在Pipeline上生成一个PCollection，输入数据。

3.对PCollection应用Transforms操做，好比wordCount中的count操做。

4.对最后的PCollection应用Sink，输出结果到外部存储中。

5.Run Pipeline到底层的计算引擎中。

使用Beam实现WordCount代码样例

public static class CountWords extends PTransform<PCollection<String>,

    PCollection<KV<String, Long>>> {

  @Override

  public PCollection<KV<String, Long>> apply(PCollection<String> lines) {

    // Convert lines of text into individual words.

    PCollection<String> words = lines.apply(

        ParDo.of(new ExtractWordsFn()));

    // Count the number of times each word occurs.

    PCollection<KV<String, Long>> wordCounts =

        words.apply(Count.<String>perElement());

    return wordCounts;

  }

}

借助Beam，用户能够利用高性能的Galaxy引擎，定制面向特定领域的系统交互接口。同时，Galaxy从此也将兼容更多生态（如Spark Streaming和Flink Streaming API）。

 

2.6 可视化集成开发平台和自动化运维

Galaxy还提供了“一站式”的集成开发环境——贝叶斯（Bayes，https://data.aliyun.com/product/sc）和自动化运维平台——特斯拉（Tesla）。经过它们，用户能够方便地管理流计算应用的生命周期，包括编程、调试、监控运维，极大地下降了流计算系统的使用门槛。

图：贝叶斯集成开发环境

 

2.7 双11的宝贵工程经验！

为保障系统在双11平稳支撑业务，在以上功能基础上，咱们还总结了完整的全链路保障方法：

• 主备双链路容灾：利用Galaxy对多副本执行的支持，面向双11重点媒体大屏等实时业务，实现了跨机房的多链路副本。哪怕是整个机房的故障，都能在秒级自动切换到另外一副本上执行，保障了双11系统高可用。
• 实时全链路监控：咱们从数据采集、读取、消费、入库各个环节都增长延时指标的埋点，能够清晰地看到整条链路各个阶段的延时，快速分析哪一个组件性能瓶颈。另外，针对做业自己运行状况，好比输入吞吐、流量、CPU和内存消耗，都作了实时分析和展现的系统，能在秒级发现做业的异常。
• 运维诊断工具：为应对各类应急响应，咱们作了一套完整的运维诊断工具用于发现集群热点机器、热点做业。在Tesla页面上能快速找到集群的热点机器，经过“机器分析”工具查看这台机器上实时跑的任务，而且能定位到相应的业务和用户。经过“做业分析”工具能自动诊断异常，结合做业的优先级，实现了一键负载均衡、启停、续跑等运维操做。

经过这些保障设施，双11当天，即便在发生交换机硬件故障的状况下，面向全球直播的媒体大屏业务并无受到任何影响！

 

2.8 小结

拥有这些和其它诸多能力，Galaxy已经具有了至关完善的实时计算能力，也提供了“一站式”的解决方案。今年双11当天，Galaxy处理了PB级别数据，处理峰值达到了1亿事件每秒，平均处理延迟在毫秒级！除了双11媒体大屏，Galaxy还支撑着阿里巴巴集团内外众多实时业务，包括数据运营、广告营销、搜索个性化、智能客服、物流调度、支付宝、聚划算等。

 

3. MaxCompute

每一年双11都是阿里巴巴从最“前端”到最“后台”全部系统整条链路的一次大考。电商在线系统的浏览和消费产生了大量数据，其数据量是日常的数倍到数十倍。这些数据最终要流到阿里巴巴的大数据计算服务—MaxCompute上来处理。

 

MaxCompute承载了阿里巴巴集团全部的离线计算任务，是集团内部核心大数据平台。截止到目前支撑着每日百万级规模的做业，整个系统拥有数万台机器，单集群规模上万，存储已经到达了EB级别，天天有数千位活跃的工程师在平台上作数据处理。

 

面对如此多的海量数据，首先须要可以低成本的将数据存储下来。MaxCompute依托背后的飞天分布式操做系统，将大量低成本PC服务管理起来。早在2013年，咱们基于对业务增加速度的判断，发现系统的存储立刻就要“撞墙”了，集群的规模将要应付不了与日俱增的数据量。直到后来成立了5k项目组，对技术难点进行了攻坚，将单集群规模扩大到了5000台，阿里巴巴也成为了中国首个独立研发拥有大规模通用计算平台技术的公司。

 

实际上单集群规模到达上万台自己技术挑战很是大，由于规模上来之后对系统设计要求很是高，整个架构不能有单点。可是整个业务规模决定了1万台机器是不够的，所以MaxCompute抽象出来一个控制层，将分布在各个不一样数据中心的多个计算集群统一管理，根据业务特色将不一样的业务放在不一样的计算集群中，经过跨集群复制，自动将数据在多个集群中同步，使得用户能够把计算引擎当成一个平台。

 

3.1 跨集群复制和全局调度

运行在MaxCompute上的业务种类很是多，各个业务部门之间数据也有着错综复杂的依赖关系。若是刚好数据不在同一个地域/机房中，那么就要进行数据的异地读写。好比分析支付宝的数据须要淘宝的数据，支付宝的数据和淘宝的数据并不在同一个机房集群，那就须要跨集群的去读（直读），或者将数据拷贝到本地再读（跨集群复制）。此外因为数据是会被更新的，好比淘宝的数据更新了，这个时候要求支付宝的做业可以读到最新版本的数据。生产任务有各自的基线时间，对处理时间有要求，不能因为互访数据致使任务延时太长。机房之间虽然有几十到上百G的直连网络专线，但其余生产业务也对网络带宽有需求，互访数据不能把带宽都占满，须要有网络流量控制。多个任务可能会访问同一份异地数据，再考虑带宽占用的限制，因此访问异地数据不能所有都经过直读异地数据来解决，有的异地数据须要在本地复制一份以供屡次任务使用。

 

为了解决这个问题，MaxCompute引入了跨集群复制和全局调度机制。MaxCompute上全部的数据表和分区的元数据引入了版本号，当数据被更新时，其对应的计算集群版本号也会更新。版本更新后，新版本所在的计算集群的数据须要被复制到其余计算集群。但这个复制操做该什么时候发生，须要考虑多种因素，好比任务完成时效要求，多集群之间的带宽大小等。对这些因素进行全局分析，才能利用动态预先调整，远程读，复制等多种手段作到全局调度。但这一全局分析须要系统运行数据才能进行。MaxCompute中的元数据、数据血缘关系的分析，以及整个系统运行过程当中产生的数据都会收集到元数据仓库，这样能够利用平台自己的数据分析能力来分析这些数据。这些数据被用来辅助MaxCompute平台的工程师作数据化运营，甚至用来帮助系统自身进行优化。

 

3.2 基于历史运行信息的优化

经过对天天运行的做业进行分析，咱们发现大部分做业都是重复执行的。这是数据仓库中的一个典型的使用场景： 天天产生的新数据被同一套数据处理任务批量重复执行。这样的场景带来了巨大的优化机会。首先天天运行的任务所占用的资源信息会被记录下来，好比运行时占用的CPU、内存和运行时间。工程师新开发的做业在第一次运行时，申请的CPU和内存通常都会和实际占用的CPU、内存有所差异。若是申请的大于实际占用的，会形成调度的时候为做业多留资源，形成资源浪费，即资源的利用率降低。若是申请的小于实际占用的，会形成一台机器上调度的做业超过了机器可以承载的负荷。这两种资源错配的后果都会下降系统使用效率。最理想的结果是做业申请的资源与实际使用的可以彻底匹配。

 

HBO( History-ed Based Optimization) 基于历史运行信息的优化就是经过收集做业的历史运行记录，根据实际CPU、内存占用来指导做业合理设置的一种优化手段。它是对集群资源分配的一种优化，归纳起来就是根据：任务执行历史+集群状态信息+优化规则，获得最优的做业资源配置。

 

HBO包含两部分工做：

• 在线部分(Online)：查找是否存在相应的hbo优化计划，若是有，则按照计划进行资源分配并执行
• 离线部分(Offline)：从元数据仓库和神农获取任务的历史执行记录，按照必定的策略生成hbo优化计划

 

下图为HBO的流程架构图：

正常状况下，这种基于历史的优化效果很是显著，由于做业整体数据量在天与天之间变化通常不会很大。但到了双11，因为当天产生的数据量一般是前几天的数倍甚至数十倍，对于一些极限状况须要作特殊处理。好比做业instance数会由于处理的数据量增大同步增加而超过单个做业instance数量上限。依托HBO的工做，能够识别重复的做业、而且可以精准的对单个做业进行设置。利用这个能力，咱们能够在节日前先对全部做业作一次分析，好比找出输入表在去年双11当天数据量显著增涨的做业，或者找出instance数量已经快要接近极限的做业，将他们单个instance处理的数据量设大，顺利度过双11的考验。以一样的手法能够指导制做针对双11的预案，好比调整CPU、内存的设置、提早发现数据倾斜等等。