本文是美图互联网技术沙龙第十一期嘉宾分享的内容，微信官方账号回复“美图大数据平台”后台获取PPT。
如今大数据在各行业的应用越来越广泛：运营基于数据注重运营结果，产品基于数据分析注重转化率，开发基于数据衡量系统优化结果。美图有美拍、美图秀秀、梅艳相机等十几个app每个app都会做个性化推荐、搜索、报告分析、反作弊、广告等。基于数据，所以整体业务对数据的需求更多，应用更广泛。
所以美图数据技术团队的业务背景主要体现在：业务线多，应用广。这也是我们建立数据平台由业务驱动.的主要原因之一
图1
举几个美图的数据应用案例。如图1，左一为美图自研数据可视化平台DataFace，支持业务方自由拖拽生成可视化报表，方便高效的数据上报和后续分析；第二个是美拍APP首页，流行的个性化推荐。基于使用的行为数据，它推荐用户可能喜欢和感兴趣的视频列表；第三种是基于用户的作弊数据，按照一定的模型和策略进行反作弊，从而有效判断和过滤用户的作弊行为。此外，它还广泛应用于搜索、a/b实验、频道跟踪和广告。
目前美图每月有5亿活跃用户，这些用户每天产生近200亿条行为数据。整体量级还是比较大的，集群机达到千量级，PB级的历史总数据。
美图有很多业务线，每个业务线在数据上的应用都比较广泛，整体用户规模比较大。所有这些因素都促使我们建立相应的数据平台来驱动这些业务的增长，并更有效地使用数据。
是我们数据平台的整体架构，如图2所示。在数据采集部分，我们构建了采集服务器日志系统Arachnia，支持各app集成的客户端SDK，负责采集app客户端数据；同时还有基于DataX的数据集成(导入导出)；Mor爬虫平台支持抓取公网数据的可配置任务开发。
图2
数据层主要根据业务特点选择不同的存储方案。目前主要使用HDFS、MongoDB、Hbase、ES等。在数据计算部分，目前离线计算主要基于HiveMR，实时流计算是Storm、Flink和另一个自研位图系统Naix。
在数据开发领域，我们搭建了数据工场、数据总线分发、任务调度等一整套平台。数据可视化及应用部分主要基于用户需求搭建一系列数据应用平台，包括：A/B实验平台、渠道推广跟踪平台、数据可视化平台、用户画像等。
图2的右侧显示了所有组件可能依赖的一些基本服务，包括地理位置、元数据管理、唯一设备标识等。
图3显示了基本的数据架构流程图，一个典型的lamda架构。从左端的数据源集合中，Arachnia和AppSDK分别向代理服务收集器报告服务器和客户端的数据。通过分析数据协议，将数据写入kafka，然后通过一层数据分发实时流，最后业务消费kafka数据进行实时计算。
图3
离线ETL服务将负责将Kafka dump数据传输到HDFS，然后异构数据源(如MySQL、Hbase等。)会主要基于DataX和Sqoop导入导出数据，最后通过hive、kylin、spark等计算将数据写入各个存储层。最后通过统一的外部API连接业务系统和我们自己的可视化平台。
企业级数据平台的建设主要分为三个阶段：
刚开始基本都是用免费的第三方平台。这个阶段的特点是能够快速集成和看到app的一些统计指标，但缺点也很明显。除了第三方提供的那些基本指标，没有原始数据无法实现其他的分析和推荐。所以有一个从0到1的过程，让我们有自己的数据可以用；数据有了之后，因为业务线和需求的爆炸，需要提高开发效率，让更多的人参与到数据的开发和使用中，而不仅仅局限于数据研发& ampd人员。所以涉及到把数据和计算存储能力开放给所有业务线，而不是握在自己手里；数据打开，业务端会问数据任务能不能跑得更快，秒出，更实时；另一方面，为了满足业务需求集
群的规模越来越大，因此会开始考虑满足业务的同时，如何实现更节省资源。美图现在是处于第二与第三阶段的过渡期，在不断完善数据开放的同时，也逐步提升查询分析效率，以及开始考虑如何进行优化成本。接下来会重点介绍 0 到 1 以及数据开放这两个阶段我们平台的实践以及优化思路。从 0 到 1
从 0 到 1 解决从数据采集到最终可以使用数据。如图 4 所示是数据收集的演进过程，从刚开始使用类似 umeng、flurry 这类的免费第三方平台，到后面快速使用 rsync 同步日志到一台服务器上存储、计算，再到后面快速开发了一个简单的python脚本支持业务服务器上报日志，最终我们开发了服务端日志采集系统 Arachnia 以及客户端 AppSDK。
图 4
数据采集是数据的源头，在整个数据链路中是相对重要的环节，需要更多关注：数据是否完整、数据是否支持实时上报、数据埋点是否规范准确、以及维护管理成本。因此我们的日志采集系统需要满足以下需求：
能集成管理维护，包括 Agent 能自动化部署安装升级卸载、配置热更、延迟方面的监控；在可靠性方面至少需要保证 at least once；美图现在有多 IDC 的情况，需要能支持多个 IDC 数据采集汇总到数据中心；在资源消耗方面尽量小，尽量做到不影响业务。基于以上需求我们没有使用 flume、scribe、fluentd，最终选择自己开发一套采集系统 Arachnia。
图 5
图 5 是 Arachnia 的简易架构图，它通过系统大脑进行集中式管理。puppet 模块主要作为单个 IDC 内统一汇总 Agent 的 metrics，中转转发的 metrics 或者配置热更命令。采集器 Agent 主要是运维平台负责安装、启动后从 brain 拉取到配置，并开始采集上报数据到 collector。
接着看 Arachnia 的实践优化，首先是 at least once 的可靠性保证。不少的系统都是采用把上报失败的数据通过 WAL 的方式记录下来，重试再上报，以免上报失败丢失。我们的实践是去掉 WAL，增加了 coordinator 来统一的分发管理 tx 状态。
图 6
开始采集前会从 coordinator 发出 txid，source 接收到信号后开始采集，并交由 sink 发送数据，发送后会ack tx，告诉 coordinator 已经 commit。coordinator 会进行校验确认，然后再发送 commit 的信号给 source、sink 更新状态，最终 tx 完 source 会更新采集进度到持久层（默认是本地 file）。该方式如果在前面 3 步有问题，则数据没有发送成功，不会重复执行；如果后面 4 个步骤失败，则数据会重复，该 tx 会被重放。
基于上文的 at least once 可靠性保证，有些业务方是需要唯一性的，我们这边支持为每条日志生成唯一 ID 标识。另外一个数据采集系统的主要实践是：唯一定位一个文件以及给每条日志做唯一的 MsgID，方便业务方可以基于 MsgID 在发生日志重复时能在后面做清洗。
我们一开始是使用 filename，后面发现 filename 很多业务方都会变更，所以改为 inode，但是 inode linux 会回收重复利用，最后是以 inode & 文件头部内容做 hash 来作为fileID。而 MsgID 是通过 agentID & fileID & offset 来唯一确认。
数据上报之后由 collector 负责解析协议推送数据到 Kafka，那么 Kafka 如何落地到 HDFS 呢？ 首先看美图的诉求：
支持分布式处理；涉及到较多业务线因此有多种数据格式，所以需要支持多种数据格式的序列化，包括 json、avro、特殊分隔符等；支持因为机器故障、服务问题等导致的数据落地失败重跑，而且需要能有比较快速的重跑能力，因为一旦这块故障，会影响到后续各个业务线的数据使用；支持可配置的 HDFS 分区策略，能支持各个业务线相对灵活的、不一样的分区配置；支持一些特殊的业务逻辑处理，包括：数据校验、过期过滤、测试数据过滤、注入等；基于上述诉求痛点，美图从 Kafka 落地到 HDFS 的数据服务实现方式如图 7 所示。
图 7
基于 Kafka 和 MR 的特点，针对每个 kafka topic 的 partition，组装 mapper 的 inputsplit，然后起一个 mapper 进程处理消费这个批次的 kafka 数据，经过数据解析、业务逻辑处理、校验过滤、最终根据分区规则落地写到目标 HDFS 文件。落地成功后会把这次处理的 meta 信息（包括 topic、partition、开始的 offset、结束的offset）存储到 MySQL。下次再处理的时候，会从上次处理的结束的 offset 开始读取消息，开始新一批的数据消费落地。
实现了基本功能后难免会遇到一些问题，比如不同的业务 topic 的数据量级是不一样的，这样会导致一次任务需要等待 partition 数据量最多以及处理时间最长的 mapper 结束，才能结束整个任务。那我们怎么解决这个问题呢？系统设计中有个不成文原则是：分久必合、合久必分，针对数据倾斜的问题我们采用了类似的思路。
图 8
首先对数据量级较小的 partition 合并到一个 inputsplit，达到一个 mapper 可以处理多个业务的 partition 数据，最终落地写多份文件。
图 9
另外对数据量级较大的 partition 支持分段拆分，平分到多个 mapper 处理同一个 partition，这样就实现了更均衡的 mapper 处理，能更好地应对业务量级的突增。
除了数据倾斜的问题，还出现各种原因导致数据 dump 到 HDFS 失败的情况，比如因为 kafka 磁盘问题、hadoop 集群节点宕机、网络故障、外部访问权限等导致该 ETL 程序出现异常，最终可能导致因为未 close HDFS 文件导致文件损坏等，需要重跑数据。那我们的数据时间分区基本都是以天为单位，用原来的方式可能会导致一个天粒度的文件损坏，解析无法读取。
图 10
我们采用了分两阶段处理的方式：mapper 1 先把数据写到一个临时目录，mapper 2 把 Hdfs 的临时目录的数据 append 到目标文件。这样当 mapper1 失败的时候可以直接重跑这个批次，而不用重跑整天的数据；当 mapper2 失败的时候能直接从临时目录 merge 数据替换最终文件，减少了重新 ETL 天粒度的过程。
在数据的实时分发订阅写入到 kafka1 的数据基本是每个业务的全量数据，但是针对需求方大部分业务都只关注某个事件、某小类别的数据，而不是任何业务都消费全量数据做处理，所以我们增加了一个实时分发 Databus 来解决这个问题。
图 11
Databus 支持业务方自定义分发 rules 往下游的 kafka 集群写数据，方便业务方订阅处理自己想要的数据，并且支持更小粒度的数据重复利用。
图 12
图 12 可以看出 Databus 的实现方式，它的主体基于 Storm 实现了 databus topology。Databus 有两个 spout，一个支持拉取全量以及新增的 rules，然后更新到下游的分发 bolt 更新缓存规则，另外一个是从 kafka 消费的 spout。而 distributionbolt 主要是负责解析数据、规则 match，以及把数据往下游的 kafka 集群发送。
数据开放
有了原始数据并且能做离线、实时的数据开发以后，随之而来的是数据开发需求的井喷，数据研发团队应接不暇。所以我们通过数据平台的方式开放数据计算、存储能力，赋予业务方有数据开发的能力。
对实现元数据管理、任务调度、数据集成、DAG 任务编排、可视化等不一一赘述，主要介绍数据开放后，美图对稳定性方面的实践心得。
数据开放和系统稳定性是相爱相杀的关系：一方面，开放了之后不再是有数据基础的研发人员来做，经常会遇到提交非法、高资源消耗等问题的数据任务，给底层的计算、存储集群的稳定性造成了比较大的困扰；另外一方面，其实也是因为数据开放，才不断推进我们必须提高系统稳定性。
针对不少的高资源、非法的任务，我们首先考虑能否在 HiveSQL 层面能做一些校验、限制。如图 13 所示是 HiveSQL 的整个解析编译为可执行的 MR 的过程：
图 13
首先基于 Antlr 做语法的解析，生成 AST，接着做语义解析，基于AST 会生成 JAVA 对象 QueryBlock。基于 QueryBlock 生成逻辑计划后做逻辑优化，最后生成物理计划，进行物理优化后，最终转换为一个可执行的 MR 任务。
我们主要在语义解析阶段生成 QueryBlock 后，拿到它做了不少的语句校验，包括：非法操作、查询条件限制、高资源消耗校验判断等。
第二个在稳定性方面的实践，主要是对集群的优化，包括：
我们完整地对 Hive、Hadoop 集群做了一次升级。主要是因为之前在低版本有 fix 一些问题以及合并一些社区的 patch，在后面新版本都有修复；另外一个原因是新版本的特性以及性能方面的优化。我们把 Hive 从 0.13 版本升级到 2.1 版本，Hadoop 从 2.4 升级到 2.7；对 Hive 做了 HA 的部署优化。我们把 HiveServer 和 MetaStoreServer 拆分开来分别部署了多个节点，避免合并在一个服务部署运行相互影响；之前执行引擎基本都是 On MapReduce 的，我们也在做 Hive On Spark 的迁移，逐步把线上任务从 Hive On MR 切换到 Hive On Spark；拉一个内部分支对平时遇到的一些问题做 bugfix 或合并社区 patch 的特性；在平台稳定性方面的实践最后一部分是提高权限、安全性，防止对集群、数据的非法访问、攻击等。提高权限主要分两块：API 访问与集群。
图 14
API Server ：上文提到我们有 OneDataAPI，提供给各个业务系统访问数据的统一 API。这方面主要是额外实现了一个统一认证 CA 服务，业务系统必须接入 CA 拿到 token 后来访问OneDataAPI，OneDataAPI 在 CA 验证过后，合法的才允许真正访问数据，从而防止业务系统可以任意访问所有数据指标。集群：目前主要是基于 Apache Ranger 来统一各类集群，包括 Kafka、Hbase、Hadoop 等做集群的授权管理和维护；以上就是美图在搭建完数据平台并开放给各个业务线使用后，对平台稳定性做的一些实践和优化。
那接下来对数据平台建设过程做一个简单的总结。
首先在搭建数据平台之前，一定要先了解业务，看业务的整体体量是否比较大、业务线是否比较广、需求量是否多到严重影响我们的生产力。如果都是肯定答案，那可以考虑尽快搭建数据平台，以更高效、快速提高数据的开发应用效率。如果本身的业务量级、需求不多，不一定非得套大数据或者搭建多么完善的数据平台，以快速满足支撑业务优先。在平台建设过程中，需要重点关注数据质量、平台的稳定性，比如关注数据源采集的完整性、时效性、设备的唯一标识，多在平台的稳定性方面做优化和实践，为业务方提供一个稳定可靠的平台。在提高分析决策效率以及规模逐渐扩大后需要对成本、资源做一些优化和思考。
