漫谈芒果TV数据平台构建

芒果TV作为一家视频行业前5的视频网站，目前拥有日活用户4100W，日均VV超过2个亿，其中PC 端月覆盖人数在1.57亿，手机APP的安装激活量达3.37亿，互联网电视终端激活用户3450W，IPTV全国覆盖用户1670W。我们每天的数据量约3-8TB不等。日处理日志条数在60-100亿条。

我们作为芒果TV数据团队主要负责公司数据统计与分析业务。其中统计系统覆盖12个终端、10多条产品线。输出核心指标、扩展指标23个，查询维度10余个。为公司各部门提供日常的数据做为决策、运营、产品策化与分析的数据基础。分析业务则主要偏向与用户属性分析，尽可能去构建人物原型。今天分享主要从基础平台建设、数据管理、数据质量三个方面与大家进行一些探讨。

做为公司级别的数据团队，我们的目标是准确、及时。数据是否及时，会直接反应到数据基础平台的可用性。通常一个数据基础平台，基本需要具备如下能力：数据采集、收集能力，数据清洗、集成能力、数据仓库、数据计算与分析能力、数据展现与输出能力。

我们在做数据平台基础架构中经历两次重点转拆。最开始在公司业务较小的时候我们使用简单的一些脚本与数据库就能满足常用的数据需求，在14年的时候随着公司业务的发展。我们发现传统的数据统计与分析方式已经不能满足我们对数据平台的基本需求，我们开始转向基于Hadoop生态圈数据平台，选用的组件Flume/Kafka/Storm/Hadoop/Hive/Hbase等，当我们经过一年发展之后集群节点数据达到100多个节点。

虽然能满足日常的数据需求，但是我们发现在这个发展过程中，对运维成本与开发的成本都在不断升高。一个数据团队大概需要投入30%以上的人在基础平台维护与建设上面。这对于一个偏向数据业务团队是不能接受的。我们开始第二轮架构调整，我们所确定的方案是使用公有数据平台 私有数据平台的方式。从而降低数据平台成本，专注数据本身。在这两次平台基础架构调整中我们明白了数据平台是基础，需要以公司情况来选择数据平台的建设。还需要根据数据本身情况选择数据平台的建设。

数据收集、采集是数据平台的源。数据收集是否准确直接决定到后面数据的可用性。我们在数据收集、采集这块经历了从前端自主上报 Nginx的方式，过渡到现在的SDK LogServer方式。对于数据平台建设初期，简单通过与客户端约定好规范，后端采用Nginx接收日志，将文件落地。已经可以满足大部分数据采集与收集的需求了。

但是这种方式也有较多的不足之处，如，1、在多个开发团队、多种客户端的情况容易出现上报BUG，2、使用明文非压缩方式传输数据存在安全与浪费网络带宽的情况、3、规范越来越多，开发本成与测试成本都会相应增加。4、场景需求多样化，导致原有的收集、采集已不能满足相应的需求。

我们选择了开发客户端SDK与服务端的LogServer，SDK，我们采用三层设计，网络通信层、核心层、接口层，网络通信层负责网络通信协议与数据的压缩加密；核心层负责公共数据采集与离线缓存数据以免在网络不通的情况导致数据丢失。接口层负责为上层应用提供Handler为客户端提供各种事件的数据接口的定义与规范。服务端则负责下发上报地址、兼容通信协调、数据解压与数据落地。在不同的客户端 我们选择不同的通信方式还获取较高的数据传输方与协议。如在手机端使用Google Buffer的方式会比Json 获取到的数据压缩率更高。

在数据收集落地并存放到S3之后，我们会根据数据的使用场景做不同的处理，主要分为两个方面：

1. 数据统计，清洗规则简单、需要结合一些其它数据对字段进行转义且中间清洗掉的数据需要根据清洗规则做保存，因为需要对计算结算进行数据校验。需要保证每条被清洗的数据是因为什么样的规则被清洗掉，做到有源可查。

2. 面向分析业务对数据精准要求没有统计要求高，但是计算复杂，往往需要经过多次迭代计算。因此我们向面统计开发一套基于可配置的分布清洗框架，流程如图。这样计算成本降低、稳定性、与扩展性更好。而面向分析业务我们则使用AWS EMR服务。其中包括常用的Spark/Hadoop/Hive 等组件。通过EMR与S3之间的数据交互实现数据分析的业务。

我们数据仓库选用的GreenPlum做为数据仓库的基础平台，我们选择GreenPlum主要原因为：

1. 采用MPP架构、支持PB级数据存储(50PB),能满足公司最3年的数据处理需求；

2. 高性价比，相于传统计仓库成本约为1/5 ，相比于Hive,使用简单、易操作。也更容易管理。

3. 关系性数据库，在选择对存储方式的情况下，能获取较高的查询效率；

4. 高可以性，GreeenPlum主节点分为Master/StandBy,数据节点也分主从，因为数据高可用性还是比较高的；

5. 可以使用PG的客户端直接连接GreenPlum更方便：

6. 支持资源队列的控制，因此我们选择了GreenPlum做为数据仓库的基出工具。

数据统计之GreenPlum的集群评估：

1. 一个SEG至少保证1核

2. 内存尽量多，至少保证一个SEG不少于8GB内存

3. 存储空间评估

计算可用的空间

初始存储能力=硬盘大小*硬盘数

配置RAID10,格式化磁盘空间＝（初始存储能力*0.9）/2

可用磁盘空间＝格式化硬盘空间*0.7

用户数据使用空间

使用镜像: (2*用户数据) 用户数据/3=可用磁盘空间

不使用镜像: 用户数据 用户数据/3=可用磁盘空间

计算用户数据大小

平均来说，实际占用磁盘空间大小＝用户数据*1.4

页面开销：32KB页面需要200bytes

行开销：每行24bytes,‘append-only’表需要4bytes有索引开销

索引开销：

B-Tree:唯一值*(数据类型大小 24bytes)

Bitmap: （唯一值 * 行数 * 1bit * 压缩比率/8） (唯一值*32)

数据统计之GreenPlum存储方式的选择

常用的存储方式：行存储、列存储

行存储特点：数据顺序写入BLOCK中，持续写入的情况下，一条记录命中在一个块中，查询多个字段时，因为记录在一个块中命中，速度较快。查询少量字段时，也要访问整条记录，造成一定的IO浪费。行存储的压缩比有限

列存储特点：压缩比可以做到很高。当查询少量字段时，扫描的块更少，可以节约IO还能提升效率。列存储适合非常典型的OLAP应用场景，按列做较大范围的聚合分析，或者JOIN分析。

数据统计之提升查询效率

我们的预计算过程，其实就是能过我们的自己研发的OLAP框架，将数据从事实表中定时进行聚类或统计，并将结果放入到结果表中，这样用户从可视化界面上面查询时可以尽量少的扫描数据，以至提高查询效率。目前我们OLAP框架上面每日运行的JOB数约上万个。

上面介绍关于批量统计这块的流量，实时统计这块我们使用是Kafka Storm，当数据落地在服务器之后，我们使用L2K (L2K是我们自己开一个日志传输工具，使用比较简单原理和Flume一样，没有Flume重)组件将数据从日志服务器上面发送Kafka，然后跑在Storm集群上面的JOB从Kafka Pull数据并进行计算。将结果写入到MYSQL数据库中，与上在离线数据GreenPlum一起对接数据可视化工具。

其中几个需求注意的地方：

1. 合理评估数据大小；

2. 合理选择Kafak的分区数；

3. 合理组件计算拓扑。

Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。1、客户端/服务器端需要使用的内存就越多，2、文件句柄的开销，3、降低高可用性 。创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是TP和TC，单位可以是MB/s。然后假设总的目标吞吐量是TT，那么分区数 = TT / max(TP, TC)。

数据可视化这块，我们输出场景

1. 面向产品运营的报表系统数据魔方；

2. 面向分析交式数据查询工具；

3. 面向技术的状态监控；

4. 面向决策的大屏输出。这块我们也还慢慢试尝中，向大家推荐一个开源BI工具MetaBase，是个不错的可视化工具。

这里的数据管理并不是指广意的数据管理，仅指做为一个统计系统所需要的一些数据治理的策略，主要包括目前几方面：

1. 采集数据管理

2. 数据集成流程及规则管理

3. 数据仓库元数据管理

4. 数据指标定义

5. 主数据归档管理

数据采集管理，通常我们需要明确上报方式（使用的协议与格式），上报范围（数据采集的应用范围），上报时机（触发上报事件的时间点），采集内容（采集那些字段）

数据仓库元数据管理，首先我们需要知道什么是元数据，元数据：元数据是关于数据的数据。只要有程序和数据, 元数据就是信息处理环境的一部分。但是在数据仓库中,元数据扮演一个新的重要角色。也 正因为有了元数据,可以最有效地利用数据仓库。元数据使得最终用户 / D S S分 析 员 能 够 探 索 各种可能性。

我们在做元数据管理过程一般从如下几个方面进行管理：

1. 程序员所知的数据结构。

2. 数据仓库的源数据。

3. 数据加入数据仓库时的转换。

4. 数据模型（星型或雪花型）。

我们在选择数据模型时选择是星型，星型模型因为数据的冗余所以很多统计查询不需要做外部的连接，因此一般情况下效率比雪花型模型要高。

数据指标定义，当数据进入数据仓库之后那我们需要是根据指标的定义与维度去计算相应的数值。统计指标定义需要有统计口径、统计维度、统计周期这几个方面的定义。

上面提到一个数据平台和对数据平台里面的一些数据治理的策略，最后讲一下对数据数据质量的四个要求：

完整性：完整性指的是数据信息是否存在缺失的状况，数据缺失的情况可能是整个数据记录缺失，也可能是数据中某个字段信息的记录缺失。不完整的数据所能借鉴的价值就会大大降低，也是数据质量最为基础的一项评估标准。

一致性：一致性是指数据是否遵循了统一的规范，数据集合是否保持了统一的格式。例如手机号码一定是11位的数字，IP地址一定是由 4个0到255间的数字加上”.”组成的。

准确性：准确性是指数据记录的信息是否存在异常或错误。和一致性不一样，存在准确性问题的数据不仅仅只是规则上的不一致。最为常见的数据准确性错误就如乱码。

及时性：及时性是指数据从产生到可以查看的时间间隔，也叫数据的延时时长。及时性对于数据分析本身要求并不高，但如果数据分析周期加上数据建立的时间过长，就可能导致分析得出的结论失去了借鉴意义。

自动化抽样校验

为提高数据质量，除非在数据处理中，严格按相应的规范处理数据，分析错误日志，同时也需要针对每天的数据结果，进行抽样的数据校验。可以通过脚本或相关工具，实现抽样的数据校验

吴红：芒果TV做为国内前五的视频平台，主要盈利模式为广告 会员。广告为主要收入来源，会员辅助收入来源。而广告收入由广告的订单、单价、库存等因素决定。而广告的库存又由我们平台的播放量(VV)与页面访问次数(PV)决定。会员收入同样由我们平台会员用户量与影片单点量决定。而会员用户量与单点量又由平台产品、内容、会员量播放量决定(会员的VV)决定。而一个平台播放量如何，又会受平台活跃用户(DAU),视频播放用户(UV)的影响。同样做为一个视频平台内容好坏也决定着视频播放量，而内容的好坏一般可以通过观看时长可以反馈出来。

因此我们将VV、PV、UV、DAU、时长做为基础核心指标，并根据这些核心指标我们又可以扩展出将多的扩展指标。基础核心指标 扩展指标共同形成一套数据体系，而将这些数据通过产品形态反应出来的产品的我们称之为数据魔方即问题中的数据墙。那些同理对于每条产品线、第个运营渠道，首先需要了解到的是该产品线的本质是什么，并总结出对接的量化指标，那么数据就能与业务联系起来。

吴红：数据实时处理我们流程与离线一样，当日志数据落地后，首先会经过我们的L2K组件，将需要参与实时计算的数据传输到Kafka队列里面。那个这个地方就上面提到的需要测试Kafka的吞吐量去建好Kafka里面的TOPIC及分配好分区数，根据数据量的情况部署好Kafka集群，当然少不了监控工具。我们使用的Kafka Eagle ，当然我们自己做一定的改进。Storm上面我们有两种JOB,一种是负责清洗的JOB把数据按清洗规则进行清洗然后输入到Kafka中，另外一种JOB接收清洗后的数据对数据进行指标指标，将计算结果存到MYSQL,另外L2k组件，会将清洗后结果落地一份输出到S3上面，供后面的离线计算使用。

本文转载于微信公众号: 大数据杂谈(BigdataTina2016)，更多微信文章请扫描关注公众号: