|
摘要：微店是全球领先的移动电商网络，创造了一个便利的手机购物环境，目前有超过3000万的店主使用微店销售商品。微店大数据架构师王锋，将重点描述大数据处理平台中数据采集、传输、存储、分析过程中的公共基础技术部分。
“人类正从IT时代走向DT时代”，2014年三月在北京举行的一场大数据产业推介会上，阿里巴巴集团创始人马云在主题演讲中发表了他的这一观点。这个观念提法很快就被广泛传播开来，并被人们所接受。这里笔者不准备大谈DT时代，但是相信DT时代一定是以数据处理为核心的，因此大数据技术在这里有至关重要的地位，很有幸笔者及各位看官正在这个领域努力。
曾看到一篇文章，里面有个观点，“DT时代的骨骼——大数据处理平台”，反映了大数据处理平台在互联网或者移动互联网公司的重要性。大数据处理平台其实包含了整个大数据处理过程，它承载了从数据采集、传输、存储、分析挖掘（离线 OR、实时 OR、即席查询）、可视化、价值体现的整体流程。这些在大的互联网公司，尤其以BAT为首，已经逐步成熟，而且价值体现不断放大。而在初创公司或者具有一定规模的创业公司，大数据处理平台的基础设施或开始搭建，或处于较初始的状态，或者在逐步规范中。可能有人会有另外的想法：我们公司规模没有那么大，有必要整这么一套么？是的，如果数据量很小，每天新增数据（比如应用日志）都是MB级别，或者GB级别，而以后也不会有爆发式增长，也没必要太折腾。无论如何，有一个趋势非常明确，随着公司业务发展，数据量的爆发式增长，大数据处理平台的建设势在必行。
大数据处理平台建设是对数据采集、数据传输、存储、分析挖掘（离线 OR 实时 OR 即席查询）、数据展现、价值体现的整体流程梳理。微店是目前全球领先的移动电商网络（在微店生态体系，公司旗下还有口袋购物、微店全球购、微店买家版、今日半价、YouShop等5大优势平台），创造了一个便利的手机购物环境，是全球年轻人喜爱的移动购物网络。目前有超过3000万的店主使用微店销售商品，在这样的背景下，技术部门开发部署的各种应用每天需要服务巨量日志数据，这些数据既包含用户的行为特征、兴趣爱好，也包含了应用的服务质量情况，这些都是要进行深度分析发掘的数据，重要性不言而喻。基于此，负责大数据基础设施建设的我们承担起了大数据处理平台的建设任务，为业务分析部门提供公共基础支撑。接下来，本文将重点描述大数据处理平台中数据采集、传输、存储、分析过程中的公共基础技术部分。
什么是数据集
随着业务的爆发式增长，公司部署了各种各样的应用服务，新的服务也不断被开发出来。日志数据由应用服务产生，应用服务由业务开发人员开发，由业务运维人员部署维护；分析挖掘这些数据的是数据分析人员、推荐算法开发人员等等，在实际工作过程中，由于各方关注角度不同，带来很多不必要的沟通交流成本。数据集（DATASET）正是为了在数据采集、传输、存储、分析过程中，数据关联各方对目标数据有统一的称谓、同时规范数据的使用。
图1显示了数据集的一些重要属性，原则上由业务开发部门申请创建新的数据集，申请者作为数据的owner，同时标识出其所属产品线、项目、数据类型，拟采用的数据收集方式、存储方式，数据规模情况预估以及要存储的时间。其中数据类型包含www日志（access log）、应用日志、错误日志、MySQL日志等等；数据收集包括：Agent实时收集、Rsync传输、HdfsClient上传、API推送；存储方式分为：HDFS、分布式消息队列Kafka、实时数据搜索Elasticsearch、第三方存储；数据规模预估可以对要收集的数据规模进行评估，传输层及存储层是否可以承载的一个初步判断。存储时间确定该数据集保存时间，到期后由平台方对数据集统一清理。
在数据集创建后，由数据采集端采集，经由数据传输层进入数据存储层。在这个过程中，category是数据集的一个代名词。category最初是Facebook开源的scribe配置中一个很重要的属性，标识数据传输对象，这里我们沿用了这个单词，并从开始到存储落地全程被携带。
数据集的划分是很重要的一个过程，决定了数据如何传输、存储，并被如何分析处理。一般由业务部门及分析部门确定。数据集内数据格式应一致，方便进行处理。但在实际场景下，尤其创业公司，单个业务部门内数据格式也未必统一，数据散落在多个日志文件中，单个体积相对较小，而分析人员也会关注这些数据，这种情况下为了方便处理，可以将这些划分到一个数据集下，同时在采集端对数据进行标注。典型方法，如在实时采集时日志行中加入header，由文件名或者其他特征区分数据。就像万事万物有其生命规律一样，数据集也不例外。图2描述了数据集的生命周期。
数据采集层
某一天，一个分析人员兴冲冲过来，“某某某，我要分析xxx服务打出的日志，xxx服务昨天上线了，这个需求非常重要，balabalabala......”。然后我们告诉他，让业务开发部门申请个数据集吧，数据集传输过来你就可以分析了：）。
数据集在创建后，所属产品线、项目、数据类型，拟采用的数据收集方式、存储方式，数据规模情况预估以及要存储的时间一一确定。以Agent实时采集为例，数据采集流程如图3所示。
- 由业务开发部门申请数据集
-
大数据组发布DataAgent
-
业务运维人员在业务机器部署DataAgent
-
DataAgent采集数据并传输
目前大部分业务的日志数据采用这种方式采集。DataAgent基于Flume实现，自开发Flume插件Tailsource支持多数据集、多文件实时tail，DataAgent具有以下特性：
- 支持数据集(category)配置，支持同时tail多个数据文件
-
支持checkpoint，定期（默认10s）将读出的文件offset写入本地磁盘
-
开发限速模块，可配置，支持在特殊场景下的限速传输
-
支持按照文件名tail文件，同时支持根据inode文件查找
-
支持文件软连接，在软连接改变后读取源日志文件剩余内容
-
修改Flume源码支持将Event Header写入原始数据中
-
借鉴美团DualChannel，开发了我们自己的DualChannel，支持MemChannel+FileChannel。
-
支持Kafkachannel，并修改kafkachannel源码，支持将原始数据写入Kafka，对业务分析程序透明
-
Agent自维护及智能升级
-
Agent端将监控指标发到指定ganglia监控端口，统一由监控层收集，支持数据比对，并支持根据应用参数设置报警。
DataAgent采集方式具体使用Flume，何种channel由数据类型、存储方式、数据量及业务场景综合确定。根据我们的测试，单个Agent，MemoryChannel在很多场景下，都可以达到6w+/s；KafkaChannel可以到到2.5w-3w+每秒，而FileChannel最高在1w/s，有些场景下甚至在5000/s以下。对应用日志，我们需要保证数据的高可靠性传输，同时需要保证效率，所以目前大量采用tailsource+Kafkachannel方式；而访问日志主要采用tailsource+DualChannel+AVROSink方式。
一些业务数据也会采用Rsync方式（存储方式仅限于HDFS存储）：在数据集确定后，大数据组分配rsync权限，由业务运维人员使用Rsync经过中间LVS层，将数据推送到databus指定的Rsync model（由category确定），最后由自开发的HADOOPLoader组件upload到HDFS。
采集层支持API推送，一些少量数据场景下，业务端可以直接调用我们提供的数据API，将数据直接写入KAFKA。
另外支持业务端直接使用HDFSClient写入HDFS，这种方式目前主要存在于以前遗留的一些数据收集上。因为Hadoop集群使用白名单方式对写入端IP进行授权，如果存在大量的这类客户端，会严重降低数据的传输效率，同时提高了客户端的维护成本。
数据传输层
业务运维人员部署DataAgent，或者其他收集方式后，数据集进入数据传输层。图4是数据传输层的整体架构。
DataBus统一负责对数据集的中间层传输、数据流转及数据落地，数据从业务端机器发出后中间经过LVS负载均衡层，进入Databus。Databus由几部分组成，包括：
- 基于Flume的Avro数据接收层，接收Agent端AvroSink发出的数据；
-
使用KafkaChannel实时消费Kafka数据；
-
接收syslog收集方式传入的数据，如交换机日志；
-
HadoopLoader接收Rsync传入的数据写入HDFS；
-
接收API post的数据
支持的存储方式包括：
- HDFS存储集群
-
Kafka分布式消息队列
-
Elasticsearch集群
-
第三方存储
其中，数据写入Kafka的topic由数据集（或者category）唯一确定，分析开发人员在自己的kafka consumer端配置topic为category即可消费数据。
对于向Elasticsearch的写入格式化数据需求，在Databus端，我们提供了具有较强通用性的支持。基于Flume ElasticsearchSink，修改源码，支持正则及分隔符的字段切割，并可配置，将Databus传输过来的数据集原始数据，根据配置的解析方式及字段，格式化数据为结构化数据适配Elasticsearch，写入ES集群。
除访问日志及应用日志以外，Databus支持以syslog方式收集网络设备数据。交换机设备的稳定对业务服务至关重要。以前我们缺乏对交换机的监控，在6月底，我们专门对公司内各机房几乎所有交换机以syslog方式收集设备日志到Kafka，并对日志进行实时分析，发现异常及时报警。
绝大部分数据需要写入HDFS数据长时间存储。我们使用改造后Flume HdfsSink写入HDFS。原生的HdfsSink有一些缺点，我们对部分源码进行改造：
- 在我们的场景中，单个机器上多个HdfsSink进程有出现文件同名的风险，修改其源码，在目前filepath+fileprefix+时间戳+filesuffix基础上，在时间戳及filesuffix之间增加4位随机数，使用过程中没有再出现文件同名情况。
-
HdfsSink在解析filepath及fileprefix过程中使用正则matcher去匹配，并且在每个Event处理过程中都会走这个过程，效率很低（对正则解析代码段单独测试500w event，正则解析代码段耗时53s），因为我们写入HDFS时按照数据集统一存储规范写入，所以将路径解析重写优化，并增加自己的配置属性，优化后，写入HDFS效率提升40%以上（lzo压缩）。
-
写入HDFS统一使用lzo方式写入，达到一定大小或者超过配置时间进行回滚。
目前Databus写入HDFS或者Kafka配置比较繁琐，后面需要针对此进行优化。
HadoopLoader是我们自行开发的组件，用以定期扫描Rsync推送过来的本地磁盘数据集存储目录，根据统一存储规范上传至HDFS。简单流程如下：
- 对每个数据集在内存中维护一个uploadingQueue。扫描线程发现待上传文件后，验证文件是否完整（根据对应md5验证码确定），然后将此文件加入此Queue。
-
上传线程从Queue中拿要上传的文件，从本地磁盘mv到uploading目录下，并上传。
-
上传结束，将已上传文件mv到本地磁盘done目录下。同时将本次上传文件路径，所属数据集、大小、md5验证码、上传时间、HDFS路径等信息入库。
客户端使用API post数据目前还在开发验证阶段，暂时不便透漏更多。Databus支持向第三方转发，基于Flume replica策略配置实现。
数据存储及分析层
上文已经提到，数据集在Databus中支持向HDFS、Kafka、Elasticsearch写入数据。这里主要对HDFS存储及公共分析平台搭建重点介绍。
对于海量数据的分布式存储，Hadoop/HDFS已经成为事实标准，目前不仅在各大互联网公司，甚至在电信领域以及银行也都开始陆续落地。Hadoop2对比Hadoop1，无论在HA、namenode扩展性、权限控制、资源调度及分配、资源隔离等都有极大提升。目前我们使用Hadoop 2.6.0作为公司最新集群使用版本，并对已知的重要bug打了patch。
相信在很多公司，尤其是创业型公司，初期业务快速扩张，为了方便，内部存在多个集群，且集群规模可能都不是很大，各业务使用的集群版本可能也不一样，相互依赖也很少。初期的散列部署结构，可以轻松应对业务的迅速发展。随着业务的逐步发展，各个业务部门数据共享需求越来越强烈，同时数据依赖关系也越来越复杂，分析数据中集群间数据来回搬动越来越多，同时随着数据量的迅速猛增，各集群存储空间压力加大，这时集群间资源整合就越来越必要，散列的集群部署结构阻碍了数据的共享，增加了数据处理过程外的许多数据迁移环节，降低了数据处理的性能，并且不利于集群资源的最大化利用，集群管理成本太高。曾见到有个业务每天将近20个TB的数据在多个集群间来回折腾的案例（并非多机房灾备），十分典型。
在微店同样如此，单个机房内存在着若干个大大小小的集群，集群规模在几个节点到近百个节点不等，最小规模才4个节点，版本也不近相同。资源整合尤为重要，同时兼顾各业务部门的效率。为大家谋福利，才能更好的推进资源整合工作。在实际整合过程中，集群不同的业务处理类型，计算引擎，决定如何去资源整合。我们整合的原则是存储共享优先，计算类型分类，兼顾特殊业务需求。在此原则下，我们多个集群将共享统一的HDFS存储资源，解决数据来回搬运的问题，同时各个集群统一版本，方便集群管理；按照计算类型进行整合，整合后将会有：
- 公共计算集群，负责MR、Hive、Pig、Streaming作业的处理；
-
Spark集群，对内存资源需求大，专门跑Spark作业；
-
GPU集群，负责高性能计算；
-
UDC集群，专门处理领导关心的时间要求高的业务指标数据报表。
整合后，集群使用统一的HDFS集群（规模300个节点），各计算集群物理隔离，服务器类型单独配置，有利于成本节约。
存储共享后，数据的存储规范、数据安全访问、读写权限规范等亟待建立。同时需要有统一的供数据分析开发人员使用的大数据处理平台Portal，作为唯一的用户授权、元数据访问、提交并管理作业、权限申请、集群资源使用情况查询、资源限额等等功能的入口。图5是对资源整合后的数据存储及分析处理流程简图。
分析开发人员由统一Portal访问大数据基础资源，支持用户对有权限的数据集查询数据集属性信息、数据集数据；按条件查找数据集、权限申请；支持权限的精细化管理（如业务组内权限分配）；作业管理（提交、运行、停止离线OR实时分析任务、Spark作业等等）、数据流转关系；查看资源使用情况报表等等。提交的作业由作业调度中心进行调度；支持公共UDF类库。元数据管理提供对业务数据仓库元数据的共享支持。
当前情况下，存在着很多客户机（任务提交机），用来提交作业。客户机必须经过平台管理方授权才可访问集群。
分析开发人员对数据集进行分析处理，需要经过数据集或Hive库表的授权，并提交到指定的队列（由集群管理房提前建立，对分析人员透明）。主要包括：
- 客户机授权。访问Hadoop集群的服务器称为客户机，授权才能访问。
-
用户及用户组。当前账号沿用Linux的user及group；将来会使用LDAP；用户组按照业务部门或产品线划分，灵活支持业务方的权限需求。
-
数据集授权。对数据集有读/写权限才可进行相应操作（得益于hadoop2.4新增的acl特性）。
3-1. 原始数据：Owner为超级管理员，业务部门只允许有读权限；生命周期由超级管理员统一管理。
3-2. 归档数据：为老数据(>6month)，统一使用LZMA压缩，提高压缩比。
3-3. 结果数据：Owner为业务方，建议使用统一存储结构统一管理。
3-4. 用户目录：Owner为业务方，采用容量配额管理。
3-5. tmp目录：都可读写，存放临时数据，由管理方定时清理。
4. Hive服务授权。统一的Hive MetaStore服务，按照业务部门或产品线对DB及表划分权限，并配合使用HDFS授权。
5. 队列授权。按照业务组划分队列，并分配资源;支持队列嵌套。【注：Hive原生代码无法做到超级管理员角色，需要自行修改代码实现。】
监控层
大数据处理平台的最后一环无疑是监控。监控像是我们的眼睛，无时无刻盯着大数据平台的整个处理流程，当将要出现问题时触发报警，平台管理人员及时切入避免故障发生。我们统一使用Ganglia从采集端、传输层到存储层、分析层的基础资源指标、应用指标写入Ganglia，并使用Nagios进行报警。图6、图7分别是平台下各基础组件的监控布局及DataAgent端按业务分类监控。
由于时间仓促，未能有更多的时间校对，文章中难免有纰漏，欢迎看官指正。另外微店正在面临数据爆发式增长，大数据技术、Hadoop相关开发人员急缺，有志于大数据方向，并且乐于深耕的技术人，欢迎将简历砸来，邮箱地址：wangfeng@weidian.com。
作者简介：王锋。曾任职并负责新浪研发dip分析平台架构设计、开发工作，承载了新浪及微博各产品线的离线、实时等各类业务分析需求。目前任职微店大数据架构师，负责微店大数据（hadoop）基础技术架构及服务运营，并负责完成业务类及运维类指标分析需求，逐步构建微店的监控分析平台。
|
