大数平台整体架构选型

数据处理分为三大类： 

 

• 第一类是从业务的角度，细分为查询检索、数据挖掘、统计分析、深度分析，其中深度分析分为机器学习和神经网络。
• 第二类是从技术的角度，细分为Batch、SQL、流式处理、machine learning、Deep learning。
• 第三类是编程模型，细分为离线编程模型、内存编程模型、实时编程模型。

结合前文讲述的数据源特点、分类、采集方式、存储选型、数据分析、数据处理，我在这里给出一个总体的大数据平台的架构。值得注意的是，架构图中去掉了监控、资源协调、安全日志等。 

大数据平台整体架构思考 

左侧是数据源，有实时流的数据（可能是结构化、非结构化，但其特点是实时的），有离线数据，离线数据一般采用的多为ETL的工具，常见的做法是在大数据平台里使用Sqoop或Flume去同步数据，或调一些NIO的框架去读取加载，然后写到HDFS里面，当然也有一些特别的技术存储的类型，比如HAWQ就是一个支持分布式、支持事务一致性的开源数据库。 

从业务场景来看，如果我们做统计分析，就可以使用SQL或MapReduce或streaming或Spark。如果做查询检索，同步写到HDFS的同时还要考虑写到ES里。如果做数据分析，可以建一个Cube，然后再进入OLAP的场景。 

 

为了支持这么多功能，我们怎么搭建我们的数据平台的呢？

Lambda架构

Lambda架构的主要思想是将大数据系统架构为多层个层次，分别为批处理层（batchlayer）、实时处理层（speedlayer）、服务层（servinglayer）如图（C）。

理想状态下，任何数据访问都可以从表达式Query= function(alldata)开始，但是，若数据达到相当大的一个级别（例如PB），且还需要支持实时查询时，就需要耗费非常庞大的资源。一个解决方式是预运算查询函数（precomputedquery funciton）。书中将这种预运算查询函数称之为Batch View（A），这样当需要执行查询时，可以从BatchView中读取结果。这样一个预先运算好的View是可以建立索引的，因而可以支持随机读取（B）。于是系统就变成：

（A）batchview = function(all data)；

（B）query =function(batch view)。

Lambda架构组件选型

下图给出了Lambda架构中各组件在大数据生态系统中和阿里集团的常用组件。数据流存储选用不可变日志的分布式系统Kafa、TT、Metaq；BatchLayer数据集的存储选用Hadoop的HDFS或者阿里云的ODPS；BatchView的加工采用MapReduce；BatchView数据的存储采用Mysql（查询少量的最近结果数据）、Hbase（查询大量的历史结果数据）。SpeedLayer采用增量数据处理Storm、Flink；RealtimeView增量结果数据集采用内存数据库Redis。

图（H）

Lambda是一个通用框架，各模块选型不要局限于上面给出的组件，特别是view的选型。因为View是和各业务关联非常大的概念，View选择组件时要根据业务的需求，选择最合适的组件。

Lambda架构的评估

优点：

a、数据的不可变性。里面给出的数据传输模型是在初始化阶段对数据进行实例化，这样的做法是能获益良多的。能够使得大量的MapReduce工作变得有迹可循，从而便于在不同阶段进行独立调试。

 b、强调了数据的重新计算问题。在流处理中重新计算是个主要挑战，但是经常被忽视。比方说，某工作流的数据输出是由输入决定的，那么一旦代码发生改动，我们将不得不重新计算来检视变更的效度。什么情况下代码会改动呢？例如需求发生变更，计算字段需要调整或者程序发出错误，需要进行调试。

缺点：

a、Jay Kreps认为Lambda包含固有的开发和运维的复杂性。Lambda需要将所有的算法实现两次，一次是为批处理系统，另一次是为实时系统，还要求查询得到的是两个系统结果的合并。

 

由于存在以上缺点，Linkedin的Jaykreps提出了Kappa架构如图（I）：

图（I）

1、使用Kafka或其它系统来对需要重新计算的数据进行日志记录，以及提供给多个订阅者使用。例如需要重新计算30天内的数据，我们可以在Kafka中设置30天的数据保留值。

2、当需要进行重新计算时，启动流处理作业的第二个实例对之前获得的数据进行处理，之后直接把结果数据放入新的数据输出表中。

3、当作业完成时，让应用程序直接读取新的数据记录表。

4、停止历史作业，删除旧的数据输出表。

 

Kappa架构暂时未做深入了解，在此不做评价。我个人觉得，不同的数据架构有各自的优缺点，我们使用的时候只能根据应用场景，选择更合适的架构，才能扬长避短。

 

 

• flume + Hadoop

 

在具体介绍本文内容之前，先给大家看一下Hadoop业务的整体开发流程： 

从Hadoop的业务开发流程图中可以看出，在大数据的业务处理过程中，对于数据的采集是十分重要的一步，也是不可避免的一步，从而引出我们本文的主角—Flume。本文将围绕Flume的架构、Flume的应用(日志采集)进行详细的介绍。 

 

  flume的特点：
　　flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。支持在日志系统中定制各类数据发送方，用于收集数据;同时，Flume提供对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力 。
　　flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带日志数据(字节数组形式)并且携带有头信息，这些Event由Agent外部的Source生成，当Source捕获事件后会进行特定的格式化，然后Source会把事件推入(单个或多个)Channel中。你可以把Channel看作是一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。

 

• Kafka + Hbase + Phoenix

先看一下我们数据处理的主要步骤，首先是我们SDK采集数据，采集数据之后，首先把它扔到我们的消息队列里做一个基础的持久化，之后我们会有两部分，一部分是实时统计，一部分是离线统计，这两部分统计完之后会把统计结果存下来，然后提供给我们的查询服务，最后是我们外部展示界面。我们的数据平台主要基于中间的四个绿色的部分。

关于要求，对消息队列来说肯定是吞吐量一定要大，要非常好的扩展性，如果有一个消息的波峰的话要随时能够扩展，因为所有的东西都是分布式的，所以要保证节点故障不会影响我们正常的业务。

我们的实时计算目前采用的是分钟级别的实时，没有精确到秒级，离线计算需要计算速度非常快，这两部分我们当初在考虑的时候就选用了Spark，因为Spark本身既支持实时，又支持离线，而且相对于其他的实时的方案来说，像Flink或者是Storm和Samza来说，我们不需要到秒级的这种实时，我们需要的是吞吐量，所以我们选择Spark。实时部分用的是Spark streaming，离线部分用的是Spark offline的方案。

查询方案因为我们要支持多个维度的组合排序，所以我们希望支持sql，这样的话各种组合排序就可以转化成sql的group和order操作。

消息队列 -- Kafka

消息队列我们选择的是Kafka，因为在我们看来，Kafka目前是最成熟的分布式消息队列方案，而且它的性能、扩展性也非常好，而且支持容错方案，你可以通过设置冗余来保证数据的完整性。 Kafka目前得到了所有主流流式计算框架的支持，像Spark, Flink, Storm, Samza等等；另外一个就是我们公司的几个创始人都来自于LinkedIn，他们之前在LinkedIn的时候就已经用过Kafka，对Kafka非常熟，所以我们选择了Kafka。

消息时序 -- HBase

但选定Kafka之后我们发现了一个问题就是消息时序的问题。首先我们的数据采集 程中，因为不同的用户网络带宽不一样，数据可能是有延迟的，晚到的消息反而可能更早发生，而且Kafka不同的partition之间是不保证时序的。

但是我们所有的离线统计程序都是需要按时间统计的，所以我们就需要一个支持时序的数据库帮我们把数据排好序，这里我们选了HBase。我们用消息产生的时间加上我们生成消息的ID做成它唯一的row key，进行排序和索引。

SQL On HBase -- Apache Phoenix

对于sql的方案来说，我们选择的是Phoenix。选Phoenix是因为我们考虑了目前几个SQL On HBase的方案，我们发现Phoenix的效率非常好，是因为它充分的利用了HBase coprocessor的特性，在server端进行了大量的计算，所以大量减轻了client的数据压力还有计算压力。

还有就是它支持HBase的Column Family概念，比如说我们要支持40个纬度的时候我们会有一张大宽表，如果我们把所有的列都设置一个列族的话，在查询任意一个列的时候都需要把40列的数据都读出来，这样是得不偿失的，所以Phoenix支持Column Family的话，我们就可以把不同的列根据它们的相关性分成几个列族，查询的时候可能只会命中一个到两个列族，这样大大减少了读取量。

Phoenix还支持Spark的DataSource API，支持列剪枝和行过滤的功能，而且支持数据写入。什么是Spark的DataSource API呢， Spark在1.2的时候提供了DataSource API，它主要是给Spark框架提供一种快速读取外界数据的能力，这个API可以方便的把不同的数据格式通过DataSource API注册成Spark的表，然后通过Spark SQL直接读取。它可以充分利用Spark分布式的优点进行并发读取，而且Spark本身有一个很好的优化引擎，能够极大的加快Spark SQL的执行。

因为Spark最近非常的火，所以它的社区资源非常的多，基本上所有主流的框架，像我们常见的Phoenix，Cassandra, MongoDB都有Spark DataSource相关的实现。还有一个就是它提供了一个统一的数据类型，把所有的外部表都统一转化成Spark的数据类型，这样的话不同的外部表能够相互的关联和操作。

在经过上述的思考之后，我们选择了这样的一个数据框架。

首先我们最下面是三个SDK，JS、安卓和iOS，采集完数据之后会发到我们的负载均衡器，我们的负载均衡器用的是AWS，它会自动把我们这些数据发到我们的server端，server在收集完数据之后会进行一个初步的清洗，把那些不规律的数据给清洗掉，然后再把那些数据发到Kafka里，后面就进入到我们的实时和离线过程。

最终我们的数据会统计到HBase里面，对外暴露的是一个sql的接口，可以通过各种sql的组合去查询所需要的统计数据。目前我们用的主要版本，Spark用的还是1.5.1，我们自己根据我们自己的业务需求打了一些定制的patch，Hadoop用的还是2.5.2，HBase是0.98，Phoenix是4.7.0，我们修复了一些小的bug，以及加了一些自己的特性，打了自己的patch。 

 

Hadoop危机？替代HDFS的8个绝佳方案

Ceph 是一个开源、多管齐下的操作系统，因为其高性能并行文件系统的特性，有人甚至认为它是基于Hadoop环境下的HDFS的接班人，因为自2010年就有研究者在寻找这个特性。

 

Apache Flink

Apache Flink是一种可以处理批处理任务的流处理框架。该技术可将批处理数据视作具备有限边界的数据流，借此将批处理任务作为流处理的子集加以处理。为所有处理任务采取流处理为先的方法会产生一系列有趣的副作用。

这种流处理为先的方法也叫做Kappa架构，与之相对的是更加被广为人知的Lambda架构（该架构中使用批处理作为主要处理方法，使用流作为补充并提供早期未经提炼的结果）。Kappa架构中会对一切进行流处理，借此对模型进行简化，而这一切是在最近流处理引擎逐渐成熟后才可行的。

流处理模型

Flink的流处理模型在处理传入数据时会将每一项视作真正的数据流。Flink提供的DataStream API可用于处理无尽的数据流。Flink可配合使用的基本组件包括：

• Stream（流）是指在系统中流转的，永恒不变的无边界数据集
• Operator（操作方）是指针对数据流执行操作以产生其他数据流的功能
• Source（源）是指数据流进入系统的入口点
• Sink（槽）是指数据流离开Flink系统后进入到的位置，槽可以是数据库或到其他系统的连接器

为了在计算过程中遇到问题后能够恢复，流处理任务会在预定时间点创建快照。为了实现状态存储，Flink可配合多种状态后端系统使用，具体取决于所需实现的复杂度和持久性级别。

此外Flink的流处理能力还可以理解“事件时间”这一概念，这是指事件实际发生的时间，此外该功能还可以处理会话。这意味着可以通过某种有趣的方式确保执行顺序和分组。

批处理模型

Flink的批处理模型在很大程度上仅仅是对流处理模型的扩展。此时模型不再从持续流中读取数据，而是从持久存储中以流的形式读取有边界的数据集。Flink会对这些处理模型使用完全相同的运行时。

Flink可以对批处理工作负载实现一定的优化。例如由于批处理操作可通过持久存储加以支持，Flink可以不对批处理工作负载创建快照。数据依然可以恢复，但常规处理操作可以执行得更快。

另一个优化是对批处理任务进行分解，这样即可在需要的时候调用不同阶段和组件。借此Flink可以与集群的其他用户更好地共存。对任务提前进行分析使得Flink可以查看需要执行的所有操作、数据集的大小，以及下游需要执行的操作步骤，借此实现进一步的优化。

优势和局限

Flink目前是处理框架领域一个独特的技术。虽然Spark也可以执行批处理和流处理，但Spark的流处理采取的微批架构使其无法适用于很多用例。Flink流处理为先的方法可提供低延迟，高吞吐率，近乎逐项处理的能力。

Flink的很多组件是自行管理的。虽然这种做法较为罕见，但出于性能方面的原因，该技术可自行管理内存，无需依赖原生的Java垃圾回收机制。与Spark不同，待处理数据的特征发生变化后Flink无需手工优化和调整，并且该技术也可以自行处理数据分区和自动缓存等操作。

Flink会通过多种方式对工作进行分许进而优化任务。这种分析在部分程度上类似于SQL查询规划器对关系型数据库所做的优化，可针对特定任务确定最高效的实现方法。该技术还支持多阶段并行执行，同时可将受阻任务的数据集合在一起。对于迭代式任务，出于性能方面的考虑，Flink会尝试在存储数据的节点上执行相应的计算任务。此外还可进行“增量迭代”，或仅对数据中有改动的部分进行迭代。

在用户工具方面，Flink提供了基于Web的调度视图，借此可轻松管理任务并查看系统状态。用户也可以查看已提交任务的优化方案，借此了解任务最终是如何在集群中实现的。对于分析类任务，Flink提供了类似SQL的查询，图形化处理，以及机器学习库，此外还支持内存计算。

Flink能很好地与其他组件配合使用。如果配合Hadoop 堆栈使用，该技术可以很好地融入整个环境，在任何时候都只占用必要的资源。该技术可轻松地与YARN、HDFS和Kafka 集成。在兼容包的帮助下，Flink还可以运行为其他处理框架，例如Hadoop和Storm编写的任务。

目前Flink最大的局限之一在于这依然是一个非常“年幼”的项目。现实环境中该项目的大规模部署尚不如其他处理框架那么常见，对于Flink在缩放能力方面的局限目前也没有较为深入的研究。随着快速开发周期的推进和兼容包等功能的完善，当越来越多的组织开始尝试时，可能会出现越来越多的Flink部署。

总结

Flink提供了低延迟流处理，同时可支持传统的批处理任务。Flink也许最适合有极高流处理需求，并有少量批处理任务的组织。该技术可兼容原生Storm和Hadoop程序，可在YARN管理的集群上运行，因此可以很方便地进行评估。快速进展的开发工作使其值得被大家关注。

Spark or Flink or hadoop or Storm 总结

大数据系统可使用多种处理技术。

对于仅需要批处理的工作负载，如果对时间不敏感，比其他解决方案实现成本更低的Hadoop将会是一个好选择。

对于仅需要流处理的工作负载，Storm可支持更广泛的语言并实现极低延迟的处理，但默认配置可能产生重复结果并且无法保证顺序。Samza与YARN和Kafka紧密集成可提供更大灵活性，更易用的多团队使用，以及更简单的复制和状态管理。

对于混合型工作负载，Spark可提供高速批处理和微批处理模式的流处理。该技术的支持更完善，具备各种集成库和工具，可实现灵活的集成。Flink提供了真正的流处理并具备批处理能力，通过深度优化可运行针对其他平台编写的任务，提供低延迟的处理，但实际应用方面还为时过早。

最适合的解决方案主要取决于待处理数据的状态，对处理所需时间的需求，以及希望得到的结果。具体是使用全功能解决方案或主要侧重于某种项目的解决方案，这个问题需要慎重权衡。随着逐渐成熟并被广泛接受，在评估任何新出现的创新型解决方案时都需要考虑类似的问题。

数据存储: HBase vs Cassandra

	HBase	Cassandra
语言	Java	Java
出发点	BigTable	BigTable and Dynamo
License	Apache	Apache
Protocol	HTTP/REST (also Thrift)	Custom, binary (Thrift)
数据分布	表划分为多个region存在不同region server上	改进的一致性哈希（虚拟节点）
存储目标	大文件	小文件
一致性	强一致性	最终一致性，Quorum NRW策略
架构	master/slave	p2p
高可用性	NameNode是HDFS的单点故障点	P2P和去中心化设计，不会出现单点故障
伸缩性	Region Server扩容，通过将自身发布到Master，Master均匀分布Region	扩容需在Hash Ring上多个节点间调整数据分布
读写性能	数据读写定位可能要通过最多6次的网络RPC，性能较低。	数据读写定位非常快
数据冲突处理	乐观并发控制（optimistic concurrency control）	向量时钟
临时故障处理	Region Server宕机，重做HLog	数据回传机制：某节点宕机，hash到该节点的新数据自动路由到下一节点做 hinted handoff，源节点恢复后，推送回源节点。
永久故障恢复	Region Server恢复，master重新给其分配region	Merkle 哈希树，通过Gossip协议同步Merkle Tree，维护集群节点间的数据一致性
成员通信及错误检测	Zookeeper	基于Gossip
CAP	1，强一致性，0数据丢失。2，可用性低。3，扩容方便。	1，弱一致性，数据可能丢失。2，可用性高。3，扩容方便。

二：部署运维

单纯的就部署和运维hbase以及Cassandra来说，部署hbase前，需要部署的组件有zookeeper，hdfs，然后才是hbase。对应的Cassandra就比较简单很多，编译完成一个jar包，单台服务器启动一个Cassandra进程即可。

在部署hbase的时候，可能需要规划好，哪些机器跑hmaser，rs,zk,hdfs的相关进程等， 还有可能为了集群的性能，还要预先规划好多少个rs。自己人工去部署这么一个hbase集群还是比较麻烦的，更别提自己维护（阿里云ApsaraDB-HBase你值得拥有）。

Cassandra部署的时候比较简单，一个tar包搞定，由于cassandra数据落本地盘，需要人为的配置一些参数比如是否需要虚拟节点（vnode）以及多少vnode；需要基于业务的场景选择特定的key的放置策略（partitioner），这个放置策略的选择以及一些参数的配置需要一定的门槛。

简单总结下：部署运维的话，hbase依赖组件多，部署麻烦一点，但是相关资料很多，降低了难度；cassandra部署依赖少，但是配置参数多，相关资料较少。

特别是使用云HBase完全避免了部署造成的各种麻烦，比手工部署运维任何大数据数据库都方便太多。

 

 

• 预测系统架构

整体架构从上至下依次是：数据源输入层、基础数据加工层、核心业务层、数据输出层和下游系统。首先从外部数据源获取我们所需的业务数据，然后对基础数据进行加工清洗，再通过时间序列、机器学习等人工智能技术对数据进行处理分析，最后计算出预测结果并通过多种途径推送给下游系统使用。

数据源输入层：京东数据仓库中存储着我们需要的大部分业务数据，例如订单信息、商品信息、库存信息等等。而对于促销计划数据则大部分来自于采销人员通过Web系统录入的信息。除此之外还有一小部分数据通过文本形式直接上传到HDFS中。

基础数据加工层：在这一层主要通过Hive对基础数据进行一些加工清洗，去掉不需要的字段，过滤不需要的维度并清洗有问题的数据。

核心业务层：这层是系统的的核心部分，横向看又可分为三层：特征构建、预测算法和预测结果加工。纵向看是由多条业务线组成，彼此之间不发生任何交集。

• 特征构建：将之前清洗过的基础数据通过近一步的处理转化成标准格式的特征数据，提供给后续算法模型使用。

• 核心算法：利用时间序列分析、机器学习等人工智能技术进行销量、单量的预测，是预测系统中最为核心的部分。

• 预测结果加工：预测结果可能在格式和一些特殊性要求上不能满足下游系统，所以还需要根据实际情况对其进行加工处理，比如增加标准差、促销标识等额外信息。

预测结果输出层：将最终预测结果同步回京东数据仓库、MySql、HBase或制作成JSF接口供其他系统远程调用。

下游系统：包括下游任务流程、下游Web系统和其他系统。

预测系统核心介绍预测系统核心层技术选型

 

预测系统核心层技术主要分为四层：基础层、框架层、工具层和算法层。

 基础层：

HDFS用来做数据存储，Yarn用来做资源调度，BDP（Big Data Platform）是京东自己研发的大数据平台，我们主要用它来做任务调度。

 框架层：

以Spark RDD、Spark SQL、Hive为主， MapReduce程序占一小部分，是原先遗留下来的，目前正逐步替换成Spark RDD。 选择Spark除了对性能的考虑外，还考虑了Spark程序开发的高效率、多语言特性以及对机器学习算法的支持。在Spark开发语言上我们选择了Python，原因有以下三点：

• Python有很多不错的机器学习算法包可以使用，比起Spark的MLlib，算法的准确度更高。我们用GBDT做过对比，发现xgboost比MLlib里面提供的提升树模型预测准确度高出大概5%~10%。虽然直接使用Spark自带的机器学习框架会节省我们的开发成本，但预测准确度对于我们来说至关重要，每提升1%的准确度，就可能会带来成本的成倍降低。

• 我们的团队中包括开发工程师和算法工程师，对于算法工程师而言他们更擅长使用Python进行数据分析，使用Java或Scala会有不小的学习成本。

• 对比其他语言，我们发现使用Python的开发效率是最高的，并且对于一个新人，学习Python比学习其他语言更加容易。

 工具层：

一方面我们会结合自身业务有针对性的开发一些算法，另一方面我们会直接使用业界比较成熟的算法和模型，这些算法都封装在第三方Python包中。我们比较常用的包有xgboost、numpy、pandas、sklearn、scipy和hyperopt等。

• Xgboost：它是Gradient Boosting Machine的一个C++实现，xgboost最大的特点在于，它能够自动利用CPU的多线程进行并行，同时在算法上加以改进提高了精度。

• numpy：是Python的一种开源的数值计算扩展。这种工具可用来存储和处理大型矩阵，比Python自身的嵌套列表结构要高效的多（该结构也可以用来表示矩阵）。

• pandas：是基于NumPy 的一种工具，该工具是为了解决数据分析任务而创建的。Pandas 纳入了大量库和一些标准的数据模型，提供了高效地操作大型数据集所需的工具。

• sklearn：是Python重要的机器学习库，支持包括分类、回归、降维和聚类四大机器学习算法。还包含了特征提取、数据处理和模型评估三大模块。

• scipy：是在NumPy库的基础上增加了众多的数学、科学以及工程计算中常用的库函数。例如线性代数、常微分方程数值求解、信号处理、图像处理和稀疏矩阵等等。

 算法层：

我们用到的算法模型非常多，原因是京东的商品品类齐全、业务复杂，需要根据不同的情况采用不同的算法模型。我们有一个独立的系统来为算法模型与商品之间建立匹配关系，有些比较复杂的预测业务还需要使用多个模型。我们使用的算法总体上可以分为三类：时间序列、机器学习和结合业务开发的一些独有的算法。

1. 机器学习算法主要包括GBDT、LASSO和RNN ：

GBDT：是一种迭代的决策树算法，该算法由多棵决策树组成，所有树的结论累加起来做最终答案。我们用它来预测高销量，但历史规律不明显的商品。

RNN：这种网络的内部状态可以展示动态时序行为。不同于前馈神经网络的是，RNN可以利用它内部的记忆来处理任意时序的输入序列，这让它可以更容易处理如时序预测、语音识别等。

LASSO：该方法是一种压缩估计。它通过构造一个罚函数得到一个较为精炼的模型，使得它压缩一些系数，同时设定一些系数为零。因此保留了子集收缩的优点，是一种处理具有复共线性数据的有偏估计。用来预测低销量，历史数据平稳的商品效果较好。

2. 时间序列主要包括ARIMA和Holt winters ：

ARIMA：全称为自回归积分滑动平均模型，于70年代初提出的一个著名时间序列预测方法，我们用它来主要预测类似库房单量这种平稳的序列。

Holt winters：又称三次指数平滑算法，也是一个经典的时间序列算法，我们用它来预测季节性和趋势都很明显的商品。

3. 结合业务开发的独有算法包括WMAStockDT、SimilarityModel和NewProduct等：

WMAStockDT：库存决策树模型，用来预测受库存状态影响较大的商品。

SimilarityModel：相似品模型，使用指定的同类品数据来预测某商品未来销量。

NewProduct：新品模型，顾名思义就是用来预测新品的销量。

预测系统核心流程

预测核心流程主要包括两类：以机器学习算法为主的流程和以时间序列分析为主的流程。

 1. 以机器学习算法为主的流程如下：

 

• 特征构建：通过数据分析、模型试验确定主要特征，通过一系列任务生成标准格式的特征数据。

• 模型选择：不同的商品有不同的特性，所以首先会根据商品的销量高低、新品旧品、假节日敏感性等因素分配不同的算法模型。

• 特征选择：对一批特征进行筛选过滤不需要的特征，不同类型的商品特征不同。

• 样本分区：对训练数据进行分组，分成多组样本，真正训练时针对每组样本生成一个模型文件。一般是同类型商品被分成一组，比如按品类维度分组，这样做是考虑并行化以及模型的准确性。

• 模型参数：选择最优的模型参数，合适的参数将提高模型的准确度，因为需要对不同的参数组合分别进行模型训练和预测，所以这一步是非常耗费资源。

• 模型训练：待特征、模型、样本都确定好后就可以进行模型训练，训练往往会耗费很长时间，训练后会生成模型文件，存储在HDFS中。

• 模型预测：读取模型文件进行预测执行。

• 多模型择优：为了提高预测准确度，我们可能会使用多个算法模型，当每个模型的预测结果输出后系统会通过一些规则来选择一个最优的预测结果。

• 预测值异常拦截：我们发现越是复杂且不易解释的算法越容易出现极个别预测值异常偏高的情况，这种预测偏高无法结合历史数据进行解释，因此我们会通过一些规则将这些异常值拦截下来，并且用一个更加保守的数值代替。

• 模型评价：计算预测准确度，我们通常用使用mapd来作为评价指标。

• 误差分析：通过分析预测准确度得出一个误差在不同维度上的分布，以便给算法优化提供参考依据。

• 

   

 2. 以时间序列分析为主的预测流程如下：

• 生成历史时序：将历史销量、价格、库存等数据按照规定格式生成时序数据。

• 节假日因子：计算节假日与销量之间的关系，用来平滑节假日对销量影响。

• 周日因子：计算周一到周日这7天与销量的关系，用来平滑周日对销量的影响。

• 促销因子：计算促销与销量之间的关系，用来平滑促销对销量的影响。

• 因子平滑：历史销量是不稳定的，会受到节假日、促销等影响，在这种情况下进行预测有很大难度，所以需要利用之前计算的各类因子对历史数据进行平滑处理。

• 时序预测：在一个相对平稳的销量数据上通过算法进行预测。

• 因子叠加：结合未来节假日、促销计划等因素对预测结果进行调整。

Spark在预测核心层的应用

我们使用Spark SQL和Spark RDD相结合的方式来编写程序，对于一般的数据处理，我们使用Spark的方式与其他无异，但是对于模型训练、预测这些需要调用算法接口的逻辑就需要考虑一下并行化的问题了。我们平均一个训练任务在一天处理的数据量大约在500G左右，虽然数据规模不是特别的庞大，但是Python算法包提供的算法都是单进程执行。我们计算过，如果使用一台机器训练全部品类数据需要一个星期的时间，这是无法接收的，所以我们需要借助Spark这种分布式并行计算框架来将计算分摊到多个节点上实现并行化处理。

我们实现的方法很简单，首先需要在集群的每个节点上安装所需的全部Python包，然后在编写Spark程序时考虑通过某种规则将数据分区，比如按品类维度，通过groupByKey操作将数据重新分区，每一个分区是一个样本集合并进行独立的训练，以此达到并行化。流程如下图所示：

 

 

伪码如下：

 

repartitionBy方法即设置一个重分区的逻辑返回(K,V)结构RDD，train方法是训练数据，在train方法里面会调用Python算法包接口。saveAsPickleFile是Spark Python独有的一个Action操作，支持将RDD保存成序列化后的sequnceFile格式的文件，在序列化过程中会以10个一批的方式进行处理，保存模型文件非常适合。

虽然原理简单，但存在着一个难点，即以什么样的规则进行分区，key应该如何设置。为了解决这个问题我们需要考虑几个方面，第一就是哪些数据应该被聚合到一起进行训练，第二就是如何避免数据倾斜。

针对第一个问题我们做了如下几点考虑：

• 被分在一个分区的数据要有一定的相似性，这样训练的效果才会更好，比如按品类分区就是个典型例子。

• 分析商品的特性，根据特性的不同选择不同的模型，例如高销商品和低销商品的预测模型是不一样的，即使是同一模型使用的特征也可能不同，比如对促销敏感的商品就需要更多与促销相关特征，相同模型相同特征的商品应倾向于分在一个分区中。

针对第二个问题我们采用了如下的方式解决：

• 对于数据量过大的分区进行随机抽样选取。

• 对于数据量过大的分区还可以做二次拆分，比如图书小说这个品类数据量明显大于其他品类，于是就可以分析小说品类下的子品类数据量分布情况，并将子品类合并成新的几个分区。

• 对于数据量过小这种情况则需要考虑进行几个分区数据的合并处理。

总之对于后两种处理方式可以单独通过一个Spark任务定期运行，并将这种分区规则保存。

结合图解Spark进行应用、优化

注：《图解Spark：核心技术与案例实战》为本文作者所著。

《图解Spark：核心技术与案例实战》一书以Spark2.0版本为基础进行编写，系统介绍了Spark核心及其生态圈组件技术。其内容包括Spark生态圈、实战环境搭建和编程模型等，重点介绍了作业调度、容错执行、监控管理、存储管理以及运行架构，同时还介绍了Spark生态圈相关组件，包括了Spark SQL的即席查询、Spark Streaming的实时流处理、MLlib的机器学习、GraphX的图处理和Alluxio的分布式内存文件系统等。下面介绍京东预测系统如何进行资源调度，并描述如何使用Spark存储相关知识进行系统优化。

结合系统中的应用

在图解Spark书的第六章描述了Spark运行架构，介绍了Spark集群资源调度一般分为粗粒度调度和细粒度调度两种模式。粗粒度包括了独立运行模式和Mesos粗粒度运行模式，在这种情况下以整个机器作为分配单元执行作业，该模式优点是由于资源长期持有减少了资源调度的时间开销，缺点是该模式中无法感知资源使用的变化，易造成系统资源的闲置，从而造成了资源浪费。

而细粒度包括了Yarn运行模式和Mesos细粒度运行模式,该模式的优点是系统资源能够得到充分利用，缺点是该模式中每个任务都需要从管理器获取资源，调度延迟较大、开销较大。

由于京东Spark集群属于基础平台，在公司内部共享这些资源，所以集群采用的是Yarn运行模式，在这种模式下可以根据不同系统所需要的资源进行灵活的管理。在YARN-Cluster模式中，当用户向YARN集群中提交一个应用程序后，YARN集群将分两个阶段运行该应用程序：

第一个阶段是把Spark的SparkContext作为Application Master在YARN集群中先启动；第二个阶段是由Application Master创建应用程序，然后为它向Resource Manager申请资源，并启动Executor来运行任务集，同时监控它的整个运行过程，直到运行完成。下图为Yarn-Cluster运行模式执行过程：

 

结合系统的优化

我们都知道大数据处理的瓶颈在IO。我们借助Spark可以把迭代过程中的数据放在内存中，相比MapReduce写到磁盘速度提高近两个数量级；另外对于数据处理过程尽可能避免Shuffle，如果不能避免则Shuffle前尽可能过滤数据，减少Shuffle数据量；最后，就是使用高效的序列化和压缩算法。在京东预测系统主要就是围绕这些环节展开优化，相关Spark存储原理知识可以参见图解Spark书第五章的详细描述。

由于资源限制，分配给预测系统的Spark集群规模并不是很大,在有限的资源下运行Spark应用程序确实是一个考验，因为在这种情况下经常会出现诸如程序计算时间太长、找不到Executor等错误。我们通过调整参数、修改设计和修改程序逻辑三个方面进行优化：

 参数调整

• 减少num-executors，调大executor-memory，这样的目的是希望Executor有足够的内存可以使用。

• 查看日志发现没有足够的空间存储广播变量，分析是由于Cache到内存里的数据太多耗尽了内存，于是我们将Cache的级别适当调成MEMORY_ONLY_SER和DISK_ONLY。

• 针对某些任务关闭了推测机制，因为有些任务会出现暂时无法解决的数据倾斜问题，并非节点出现问题。

• 调整内存分配，对于一个Shuffle很多的任务，我们就把Cache的内存分配比例调低，同时调高Shuffle的内存比例。

 修改设计

参数的调整虽然容易做，但往往效果不好，这时候需要考虑从设计的角度去优化：

• 原先在训练数据之前会先读取历史的几个月甚至几年的数据，对这些数据进行合并、转换等一系列复杂的处理，最终生成特征数据。由于数据量庞大，任务有时会报错。经过调整后当天只处理当天数据，并将结果保存到当日分区下，训练时按天数需要读取多个分区的数据做union操作即可。

• 将“模型训练”从每天执行调整到每周执行，将“模型参数选取”从每周执行调整到每月执行。因为这两个任务都十分消耗资源，并且属于不需要频繁运行，这么做虽然准确度会略微降低，但都在可接受范围内。

• 通过拆分任务也可以很好的解决资源不够用的问题。可以横向拆分，比如原先是将100个品类数据放在一个任务中进行训练，调整后改成每10个品类提交一次Spark作业进行训练。这样虽然整体执行时间变长，但是避免了程序异常退出，保证任务可以执行成功。除了横向还可以纵向拆分，即将一个包含10个Stage的Spark任务拆分成两个任务，每个任务包含5个Stage，中间数据保存到HDFS中。

 修改程序逻辑

为了进一步提高程序的运行效率，通过修改程序的逻辑来提高性能，主要是在如下方面进行了改进：避免过多的Shuffle、减少Shuffle时需要传输的数据和处理数据倾斜问题等。

1. 避免过多的Shuffle

Spark提供了丰富的转换操作，可以使我们完成各类复杂的数据处理工作，但是也正因为如此我们在写Spark程序的时候可能会遇到一个陷阱，那就是为了使代码变的简洁过分依赖RDD的转换操作，使本来仅需一次Shuffle的过程变为了执行多次。我们就曾经犯过这样一个错误,本来可以通过一次groupByKey完成的操作却使用了两回。

业务逻辑是这样的：我们有三张表分别是销量（s）、价格（p）、库存（v），每张表有3个字段：商品id（sku_id）、品类id（category）和历史时序数据（data），现在需要按sku_id将s、p、v数据合并，然后再按category再合并一次，最终的数据格式是：[category，[[sku_id, s , p, v], [sku_id, s , p, v], […]，[…]]]。一开始我们先按照sku_id + category作为key进行一次groupByKey，将数据格式转换成[sku_id, category , [s，p, v]]，然后按category作为key再groupByKey一次。

后来我们修改为按照category作为key只进行一次groupByKey，因为一个sku_id只会属于一个category，所以后续的map转换里面只需要写一些代码将相同sku_id的s、p、v数据group到一起就可以了。两次groupByKey的情况：

 

 

 

修改后变为一次groupByKey的情况：

 

多表join时，如果key值相同，则可以使用union+groupByKey+flatMapValues形式进行。比如：需要将销量、库存、价格、促销计划和商品信息通过商品编码连接到一起，一开始使用的是join转换操作，将几个RDD彼此join在一起。后来发现这样做运行速度非常慢，于是换成union+groypByKey+flatMapValue形式，这样做只需进行一次Shuffle，这样修改后运行速度比以前快多了。实例代码如下：

 

如果两个RDD需要在groupByKey后进行join操作，可以使用cogroup转换操作代替。比如， 将历史销量数据按品类进行合并，然后再与模型文件进行join操作，流程如下：

 

使用cogroup后，经过一次Shuffle就可完成了两步操作，性能大幅提升。

2. 减少Shuffle时传输的数据量

• 在Shuffle操作前尽量将不需要的数据过滤掉。

• 使用comebineyeByKey可以高效率的实现任何复杂的聚合逻辑。

• 

   

comebineyeByKey属于聚合类操作，由于它支持map端的聚合所以比groupByKey性能好，又由于它的map端与reduce端可以设置成不一样的逻辑，所以它支持的场景比reduceByKey多，它的定义如下：

educeByKey和groupByKey内部实际是调用了comebineyeByKey，

 

 

我们之前有很多复杂的无法用reduceByKey来实现的聚合逻辑都通过groupByKey来完成的，后来全部替换为comebineyeByKey后性能提升了不少。

3.处理数据倾斜

有些时候经过一系列转换操作之后数据变得十分倾斜，在这样情况下后续的RDD计算效率会非常的糟糕，严重时程序报错。遇到这种情况通常会使用repartition这个转换操作对RDD进行重新分区，重新分区后数据会均匀分布在不同的分区中，避免了数据倾斜。如果是减少分区使用coalesce也可以达到效果，但比起repartition不足的是分配不是那么均匀。

 

 

其它阅读:

facebook为什么放弃Cassandra？

参考：http://www.zhihu.com/question/19593207:

Facebook开发Cassandra初衷是用于Inbox Search，但是后来的Message System则使用了HBase，Facebook对此给出的解释是Cassandra的最终一致性模型不适合Message System，HBase具有更简单的一致性模型，当然还有其他的原因。HBase更加的成熟，成功的案例也比较多等等。Twitter和Digg都曾经很高调的选用Cassandra，但是最后也都放弃了，当然Twitter还有部分项目也还在使用Cassandra，但是主要的Tweet已经不是了。