Spark技术在京东智能供应链预测的应用

1 背景

 前段时间京东公开了面向第二个十二年的战略规划，表示京东将全面走向技术化，大力发展人工智能和机器人自动化技术，将过去传统方式构筑的优点全面升级。京东Y事业部顺势成立，该事业部将以服务泛零售为核心，着重智能供应能力的打造，核心使命是利用人工智能技术来驱动零售革新。

1.1   京东的供应链

京东一直致力于经过互联网电商创建需求侧与供给侧的精准、高效匹配，供应链管理是零售联调中的核心能力，是零售平台能力的关键体现，也是供应商与京东紧密合做的纽带，更是将来京东智能化商业体布局中的核心环节。

目前京东在全国范围内的仓库数量已超过700个，按功能可划分为RDC、FDC、大件中心仓、大件卫星仓、图书仓和城市仓等等。RDC（Regional Distribution Center）即区域分发中心，可理解为一级仓库，向供货商采购的商品会优先送往这里，通常设置在中心城市，覆盖范围大。FDC（Forward Distribution Center）即区域运转中心，可理解为二级仓库，覆盖一些中、小型城市及边远地区，一般会根据需求将商品从RDC调配过来。

结合人工智能、大数据等技术，京东首先从供货商那里合理采购定量的商品到RDC，再根据实际需求调配到FDC，而后运往离客户最近的配送站，最后快递员将商品带到客户手中。这只是京东供应链体系中一个普通的场景，但正由于有这样的体系，使得京东对用户的响应速度大大提升，用户体验大大提高。

1.2   京东供应链优化

用户体验提高的同时也伴随着大量资金的投入和成本的提升，成本必须获得控制，整个体系才能发挥出最大的价值，因而对供应链的优化就显得相当重要了。

京东自打创建供应连体系的那一天起，就不断地进行改进和优化，而且努力深刻到供应链的每个环节。优化实际上是一门运筹学问题，需考虑在各类决策目标之间如何平衡以达到最大收益，在这个过程当中须要考虑不少问题，把这些考虑清楚，问题就容易解决了。举几个简单的例子：

l  商品补货：考虑在什么时间，给哪一个RDC采购什么商品，采购量是多少？

l  商品调拨：考虑在什么时间，给哪一个FDC调配什么商品，调配量是多少？

l  仓储运营：在大促来临之际，仓库和配送站要增配多少人手、多少辆货车？

虽然看上去这些问题都很容易回答，但仔细想一想却又很难给出答案，缘由就在于想要作到精确不是那么容易的事情，就拿补货来讲，补的太多会增长库存成本，补的太少会增长缺货成本，只有合理的补货量才能作到成本最低。

1.3   预测技术在京东供应链的做用

借助机器学习、大数据等相关技术，京东在不少供应链优化问题上都已经实现系统化，由系统自动给出优化建议，并与生产系统相链接，实现全流程自动化。在这里有一项技术起着相当重要的低层支撑做用--预测技术。据粗略估算，1%的预测准确度的提高能够节约数倍的运营成本。

怎样理解预测在供应链优化中的做用呢?拿商品补货举例，一家公司为了保证库房不缺货，可能会频繁的从供货商那里补充大量商品，这样作虽然不会缺货，但可能会形成更多卖不出去的商品积压在仓库中，从而使商品的周转率下降，库存成本增长。反之，这家公司有可能为了追求零库存而补不多的商品，但这就可能出现严重的缺货问题，从而使现货率下降，严重影响用户体验，缺货成本增长。因而问题就来了，要补多少商品才合适，什么时间补货，这就须要权衡考虑了，最终目的是要使库存成本和缺货成本达到一个平衡。

考虑一下极端状况，等库存降到零时再去补货，这时供货商接到补货通知后将货物运往仓库。可是这么作有个问题，由于运送过程须要时间，这段时间库房就缺货了。那怎么办呢?就是利用预测技术。利用预测咱们能够计算出将来商品在途的这段时间里销量大概是多少，而后咱们让仓库保证这个量，低于这个量就给供货商下达补货通知，因而问题得以解决。总而言之，预测技术在这里发挥了重要的做用，成为关键的一个环。

2 京东预测系统

2.1 预测系统介绍

预测系统在整个供应链体系中处在最底层而且起到一个支撑的做用，支持上层的多个决策优化系统，而这些决策优化系统利用精准的预测数据结合运筹学技术得出最优的决策，并将结果提供给更上层的业务执行系统或是业务方直接使用。

目前，预测系统主要支持三大业务：销量预测、单量预测和GMV预测。其中销量预测主要支持商品补货、商品调拨；单量预测主要支持仓库、站点的运营管理；GMV预测主要支持销售部门计划的定制。

销量预测按照不一样维度又能够分为RDC采购预测、FDC调拨预测、城市仓调拨预测、大建仓补货预测、全球购销量预测和图书促销预测等；单量预测又可分为库房单量预测、配送中心单量预测和配送站单量预测等（在这里“单量”并不是指用户所下订单的量，而是将订单拆单后流转到仓库中的单量。例如一个用户的订单中包括3件物品，其中两个大件品和一个小件品，在京东的供应链环节中可能会将其中两个大件品组成一个单投放到大件仓中，而将那个小件单独一个单投放到小件仓中，单量指的是拆单后的量）；GMV预测支持到商品粒度。

2.2 预测系统架构

       

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

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

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

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

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

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

Ø  预测结果加工：预测结果可能在格式和一些特殊性要求上不能知足下游系统，因此还须要根据实际状况对其进行加工处理，好比增长标准差、促销标识等额外信息。

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

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

3 预测系统核心介绍

3.1 预测系统核心层技术选型

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

基础层：

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

框架层：  

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

l  Python有不少不错的机器学习算法包可使用，比起Spark的MLlib，算法的准确度更高。咱们用GBDT作过对比，发现xgboost比MLlib里面提供的提高树模型预测准确度高出大概5%~10%。虽然直接使用Spark自带的机器学习框架会节省咱们的开发成本，但预测准确度对于咱们来讲相当重要，每提高1%的准确度，就可能会带来成本的成倍下降。

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

l  对比其余语言，咱们发现使用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：新品模型，顾名思义就是用来预测新品的销量。

3.2 预测系统核心流程

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

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

      

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

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

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

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

模型参数：选择最优的模型参数，合适的参数将提升模型的准确度，由于须要对不一样的参数组合分别进行模型训练和预测，因此这一步是很是耗费资源。

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

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

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

预测值异常拦截：咱们发现越是复杂且不易解释的算法越容易出现极个别预测值异常偏高的状况，这种预测偏高没法结合历史数据进行解释，所以咱们会经过一些规则将这些异常值拦截下来，而且用一个更加保守的数值代替。

模型评价：计算预测准确度，咱们一般用使用mapd来做为评价指标。

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

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

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

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

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

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

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

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

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

3.3 Spark在预测核心层的应用

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

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

 

伪码以下： 

sc.textFile("...").map(lambda x: repartitionBy(x)).groupByKey()
  .map(lambda x: train(x)).saveAsPickleFile("...")

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

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

针对第一个问题咱们作了以下几点考虑：

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

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

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

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

l  对于数据量过大的分区还能够作二次拆分，好比图书小说这个品类数据量明显大于其余品类，因而就能够分析小说品类下的子品类数据量分布状况，并将子品类合并成新的几个分区。

l  对于数据量太小这种状况则须要考虑进行几个分区数据的合并处理。

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

4 结合图解Spark书进行应用与优化

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

4.1 结合系统中的应用

在图解Spark书的第六章描述了Spark运行架构，介绍了Spark集群资源调度通常分为粗粒度调度和细粒度调度两种模式。粗粒度包括了独立运行模式和Mesos粗粒度运行模式，在这种状况下以整个机器做为分配单元执行做业，该模式优势是因为资源长期持有减小了资源调度的时间开销，缺点是该模式中没法感知资源使用的变化，易形成系统资源的闲置，从而形成了资源浪费。而细粒度包括了Yarn运行模式和Mesos细粒度运行模式,该模式的优势是系统资源可以获得充分利用，缺点是该模式中每一个任务都须要从管理器获取资源，调度延迟较大、开销较大。

因为京东Spark集群属于基础平台，在公司内部共享这些资源，因此集群采用的是Yarn运行模式，在这种模式下能够根据不一样系统所须要的资源进行灵活的管理。在YARN-Cluster模式中，当用户向YARN集群中提交一个应用程序后，YARN集群将分两个阶段运行该应用程序：第一个阶段是把Spark的SparkContext做为Application Master在YARN集群中先启动；第二个阶段是由Application Master建立应用程序，而后为它向Resource Manager申请资源，并启动Executor来运行任务集，同时监控它的整个运行过程，直到运行完成。下图为Yarn-Cluster运行模式执行过程：

4.2   结合系统的优化

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

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

   4.2.1 参数调整

l  减小num-executors，调大executor-memory，这样的目的是但愿Executor有足够的内存可使用。

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

l  针对某些任务关闭了推测机制，由于有些任务会出现暂时没法解决的数据倾斜问题，并不是节点出现问题。

l  调整内存分配，对于一个Shuffle不少的任务，咱们就把Cache的内存分配比例调低，同时调高Shuffle的内存比例。

  4.2.2 修改设计

参数的调整虽然容易作，但每每效果很差，这时候须要考虑从设计的角度去优化：

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

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

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

4.2.3 修改程序逻辑

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

    1 避免过多的Shuffle

l  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的状况：

l  多表join时，若是key值相同，则可使用union+groupByKey+flatMapValues形式进行。好比：须要将销量、库存、价格、促销计划和商品信息经过商品编码链接到一块儿，一开始使用的是join转换操做，将几个RDD彼此join在一块儿。后来发现这样作运行速度很是慢，因而换成union+groypByKey+flatMapValue形式，这样作只需进行一次Shuffle，这样修改后运行速度比之前快多了。

实例代码以下：

rdd1 = rdd1.mapValues(lambda x: (1, x))
rdd2 = rdd2.mapValues(lambda x: (2, x))
rdd3 = rdd3.mapValues(lambda x: (3, x))
def dispatch(seq):
    vbuf, wbuf, xbuf = [], [], []
    for (n, v) in seq:           
        if n == 1: vbuf.append(v)        
        elif n == 2: wbuf.append(v)         
        elif n == 3: xbuf.append(v)    
    return [(v, w, x) for v in vbuf for w in wbuf for x in xbuf]
new_rdd = sc.union([rdd1, rdd3, rdd3]).groupByKey().flatMapValues(lambda x: dispatch(x))

l  若是两个RDD须要在groupByKey后进行join操做，可使用cogroup转换操做代替。好比， 将历史销量数据按品类进行合并，而后再与模型文件进行join操做，流程以下：

[sku_id, category , sales] 
-> [category, [[sku_id, sales], […]]] 
-> [category, [[[sku_id, sales], […]], [model1, model2]]]

 

使用cogroup后，通过一次Shuffle就可完成了两步操做，性能大幅提高。

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

l  在Shuffle操做前尽可能将不须要的数据过滤掉。

l  使用comebineyeByKey能够高效率的实现任何复杂的聚合逻辑。

comebineyeByKey属于聚合类操做，因为它支持map端的聚合因此比groupByKey性能好，又因为它的map端与reduce端能够设置成不同的逻辑，因此它支持的场景比reduceByKey多，它的定义以下 

def combineByKey(self, createCombiner, mergeValue,mergeCombiners, numPartitions)

 reduceByKey和groupByKey内部实际是调用了comebineyeByKey

def reduceByKey(self, func, numPartitions=None):
      return self.combineByKey(lambda x: x, func, func, numPartitions)

def groupByKey(self, numPartitions=None):
     def createCombiner(x):
          return [x]
     def mergeValue(xs, x):
          xs.append(x)
          return xs
     def mergeCombiners(a, b):
          a.extend(b)
          return a
     return self.combineByKey(createCombiner, mergeValue,  mergeCombiners,numPartitions).mapValues(lambda x: ResultIterable(x))

咱们以前有不少复杂的没法用reduceByKey来实现的聚合逻辑都经过groupByKey来完成的，后来所有替换为comebineyeByKey后性能提高了很多。

     3.   处理数据倾斜

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

5  小结

虽然京东的预测系统已经稳定运行了很长一段时间，可是咱们也看到系统自己还存在着不少待改进的地方，接下来咱们会在预测准确度的提升、系统性能的优化、多业务支持的便捷性上进行改进。将来，随着大数据、人工智能技术在京东供应链管理中的使用愈来愈多，预测系统也将发挥出更大做用，对于京东预测系统的研发工做也将是充满着挑战与乐趣。

 

该文本人发表于2017-03-06 InfoQ的大数据杂谈公众号，微信号：BigdataTina2016

连接地址为 https://mp.weixin.qq.com/s/35c06LQHVsyG-dy4FgJZnA