0 引言

随着大数据与云计算技术的发展，众多大数据存储、计算框架在商业界和学术界得到广泛的应用和研究。其中，Spark计算框架由于其具有强大的计算能力、分布式可拓展性和容错能力成为新一代计算框架中研究的热点。

然而，对部署在分布式环境中的计算框架进行性能监控和分析一直被视为一个非常困难的问题^[1]。这是由于在分布式环境当中，可能导致系统性能下降的因素多种多样，例如：用户配置不当、硬件故障、资源分配不公平等。若将系统进一步部署到云环境中，那么同一主机上运行的其他虚拟机会进一步对系统性能造成影响。

目前，国内外的研究大致采用三种方法对系统进行监控：1) 基于日志的工作流分析；2) 事件因果关系追踪；3) 动态追踪。

文献[2-9]使用分析系统日志的方法定位系统中性能下降的作业。分析日志的方法能够在很大程度上降低用户的工作量，在系统故障时，也可运用这些结果来定位和排除故障；但该方法也有下述不足，由于被分析的日志或控制台记录的信息需要在分析之前指定，这就导致了记录的信息量与分析效率之间的权衡问题。也就是说，若应用执行时记录的信息过多，会导致应用程序效率下降；反之，则可能导致提供给分析引擎的数据不足，不能准确定位故障。

文献[10-20]采用事件因果关系追踪的方法。比起基于日志分析的追踪技术，因果关系追踪技术在监控系统时会追踪系统中各事件的关系。在系统发生故障时，用户最先排查到的故障原因可能只是故障的表层原因。原因追踪技术可以追踪导致故障的事物流，从而帮助用户准确定位故障的根本原因。因此使用原因追踪技术能更准确地定位故障原因。比起日志分析技术，效率更高。

文献[21-23]中使用的动态追踪方法能最大限度减少追踪时对系统性能的影响。这是因为动态追踪技术在用户不开启追踪功能时，对系统性能的影响几乎为零；而开启追踪时，用户每一次需要追踪的信息也往往只与系统中一小部分模块有关。但在动态追踪中，用户对系统代码有比较深入的了解，才能指定需要追踪的系统模块。这种方法不适合普通用户。

此外，以上监控方法都没有与云环境相结合。云平台已在商业上广泛地应用，如Microsoft Azure^[24]和Amazon EC2^[25]都是知名云平台提供商，在云环境中运行的计算框架有更好的可拓展性与可伸缩性；与此同时，云环境中计算框架的性能特点也与传统物理环境中的不尽相同。因此，以上方法不能准确地对云环境中的计算框架进行监控和分析。

对此，本文提出一种对运行在虚拟环境中Spark^[26]的性能进行监控的系统和分析的方法，并实现了一套监控与分析系统。本文中，Spark被部署在Docker^[27]虚拟化容器中以模拟云环境。Docker容器为Spark作业提供了良好的资源隔离。本监控系统通过在Spark中植入代码和读取Docker系统文件，以监控Spark作业运行时的资源消耗情况。使用植入代码的方式能够更准确地提供Spark的运行时状态信息，如正在执行的作业、执行阶段以及资源消耗情况。由于不同类型作业对资源的消耗会呈现相应的特点，本文利用这些特点，采用高斯混合模型(Gaussian Mixture Model, GMM)，对采集到的信息进行分析，并从中找出异常信息以及其对应的Spark作业。实验结果表明，本监控系统异常检测的有效性为90.2%；与此同时，其对Spark性能的影响仅为4.7%。

本文方法的优势主要有以下两方面：1) 本文方法可对作业资源消耗情况进行实时分析并向用户反馈异常作业。目前许多监控系统，如Ganglia^[28]，可以对集群中的主机进行实时监控；但由于集群可被多个用户共享，同一时间内可能有多个作业运行，因此该种方式不能获取每一个作业的资源消耗情况。另外一些研究，如文献[11]，可实时重建作业执行过程中的事件流，但其工作没有涉及对作业性能的分析与反馈。

2) 本文方法将监控与Docker容器相结合，从而更准确地获取作业每个阶段的资源消耗信息。监控软件大多针对集群中的主机节点，即使在单用户环境，操作系统中的后台进程也会影响监控的准确性。虽然目前已有针对Docker的监控软件出现，用户仍然需要手动将资源消耗情况与作业各个阶段相对应。

1 相关理论 1.1 Spark框架

Spark是一个开源的通用计算框架。由于高效的计算能力、易编程性、可拓展性和良好的容错性，Spark在越来越多的企业中得到广泛的应用^[29]。

Spark使用弹性分布式数据集(Resilient Distributed Datasets, RDD)作为内存存储机制。由于数据以RDD的形式存储在内存当中，Spark的计算性能比起上一代Hadoop框架提升了10倍以上，对部分迭代型作业的性能提升甚至在100倍以上。RDD也为Spark提供了良好的容错性能，通过对父RDD的计算，一个损坏或丢失的RDD可以被快速重建。

Spark由一个负责调度作业的Master节点和多个负责执行作业的Worker节点组成。每个Worker节点可根据自身资源数量启动一定数量的Executor。Executor是Spark中分配资源和执行任务的单位。其使用的资源数量，如CPU个数和内存用量，可由用户指定。Spark可使用YARN作为底层的资源调度器。Spark on YARN的整体构架如图 1所示。

1.2 Docker容器及优点

Docker是一个开源的Linux软件容器，能提供轻量级的虚拟化环境。Docker使用cgroup技术，能对单个操作系统上进程间的CPU、内存、网络流量、文件系统等进行隔离。相比于传统虚拟化技术，Docker的优势如下：1) Docker不需要Hypervisor，直接运行在操作系统之上，能在数秒之内启动。2) Docker在为软件提供资源隔离的同时，自身基本不消耗系统资源。同一物理主机上能同时运行的Docker实例可达上千个，极大提高了系统资源的利用率。3) Docker中可集成软件运行时需要的所有依赖库，实现了软件的“一次配置，多地运行”，增强了软件的可移植性。

在本文中，Spark的Executor运行在Docker容器中。Docker为每个Executor分配资源，并为Executor与其他系统进程之间实现资源隔离。由于Executor运行时的网络流量和磁盘读写率不能从Spark内部获取，本监控系统通过监控Docker的API文件，以获取这些性能指标。

1.3 高斯混合模型及EM求解算法

同一Spark作业在集群中多次执行，正常情况下其具有相似的资源消耗特征。例如，k-means作业会占用大量CPU资源和少量网络I/O资源。但由于Spark作业在不同配置的集群中资源消耗情况不相同，用户不能将同一套分析参数应用于不同的集群。即使在同一集群中，用户也很难对大量作业的资源消耗情况进行量化，从而进一步标记分类。此外，若要实现在线异常检测，则要求算法的实时性。为满足以上需求，选用高斯混合模型(GMM)作为异常检测方法。高斯混合模型能适应不同集群，根据历史作业自动训练参数；并且，训练后的高斯混合模型能够在O(n)时间内将作业分类以检测异常。

高斯混合模型是机器学习中常用的非监督学习算法。它使用多个高斯概率密度函数来描述变量分布，不仅能将变量分类，还能计算出变量属于每一个类别的概率。对于观察样本X={x₁, x₂, …, x_N}和由K个高斯概率密度函数组成的高斯混合模型，每个高斯概率密度函数称为一个组件(component)。样本x_i (1≤i≤N)是D维向量，它属于第j个组件的概率为：

$P({{\mathit{\boldsymbol{x}}}_i}|{\mathit{\boldsymbol{\mu }}_j},{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_j}) = \frac{{\exp [ - \frac{1}{2}{{({{\mathit{\boldsymbol{x}}}_i} - {{\mathit{\boldsymbol{\mu }}}_j})}^T}{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_j}^{ - 1}({{\mathit{\boldsymbol{x}}}_i} - {{\mathit{\boldsymbol{\mu }}}_j})]}}{{\sqrt {{{(2\pi )}^D}(|{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_j}|)} }}$

(1)

其中，μ_j与Σ_j分别为第j个组件的期望向量与协方差矩阵。

高斯混合模型需要优化的对数似然函数表示为：

$LL = \sum\limits_{i = 1}^N {\lg \left\{ {\sum\limits_{j = 1}^K {{\pi _j}P({{\mathit{\boldsymbol{x}}}_i}|{{\mathit{\boldsymbol{\mu }}}_j},{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_j})} } \right\}} $

(2)

使用EM算法可求出高斯混合模型的参数。EM算法的求解过程如下：

根据先验知识，初始化每个组件的期望μ_j和协方差矩阵Σ_j。然后重复以下E-步和M-步，直到式(2) 收敛：

E-步：

$\gamma (i,k) = \frac{{{\pi _j}P({\mathit{\boldsymbol{x}}}|{{\mathit{\boldsymbol{\mu }}}_k},{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_k})}}{{\sum\limits_{j = 1}^K {{\pi _j}p({{\mathit{\boldsymbol{x}}}_i}|{{\mathit{\boldsymbol{\mu }}}_j},{{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_j})} }}$

M-步：

${N_k} = \sum\limits_{i = 1}^N {\gamma (i,k)} \\ {\pi _k} = {N_k}/N\\ {{\mathit{\boldsymbol{\mu }}}_k} = \frac{1}{{{N_k}}}\sum\limits_{i = 1}^N {\gamma (i,k){{\mathit{\boldsymbol{x}}}_i}} \\ {{\mathit{\boldsymbol{ \boldsymbol{\varSigma} }}}_k} = \frac{1}{{{N_k}}}\sum\limits_{i = 1}^N {\gamma (i,k)({{\mathit{\boldsymbol{x}}}_i} - {{\mathit{\boldsymbol{\mu }}}_k}){{({{\mathit{\boldsymbol{x}}}_i} - {{\mathit{\boldsymbol{\mu }}}_k})}^{\rm{T}}}} $

在EM算法中，E-步估算数据由每个组件生成的概率，M-步利用E-步的结果，估算每个高斯分布函数的参数值。每次E-步和M-步迭代完成后，利用M-步的结果，可重新计算对数似然函数式(2)。

不同Spark作业的资源消耗情况虽然不尽相同，但各自也具有一定的特点，如CPU密集型作业和I/O密集型作业。利用高斯混合模型，可根据Spark作业的资源消耗情况将其分类。由于高斯混合模型中包含样本属于每一个分类的概率，异常的Spark作业可以通过设置阈值检测到。

2 监控系统设计

在本文方法中，通过在Spark的Executor模块植入代码和读取Docker文件，实现了对Spark的运行时信息的监控。使用植入代码的方式，可以高效准确地从Spark内部获取到其运行时信息，如作业、阶段的开始和结束时间等。获取到的Spark和Docker监控信息被传递给监控系统，由监控系统进行整合并存储到数据库。监控系统通过使用Spark历史作业信息训练高斯混合模型。当新作业提交时，监控系统会在作业运行的同时分析作业的性能指标，并向用户报告运行异常的作业。本文监控系统的架构如图 2所示，其中灰色部分为本文的主要工作。① 用户首先提交作业到Spark Master节点；② Master节点向运行在Work节点上的Executor分配任务；③ 植入在Executor中的监控管理器向监控系统注册任务并实时报告任务资源使用情况；④ Docker监控器收集并报告Docker资源使用情况；⑤ 整合后的信息整存储到数据库；⑥ 整合后的信息送到数据分析模块；⑦ 分析模块向用户反馈异常的作业和阶段。

2.1 Spark代码植入

在Spark中植入代码时，必须权衡代码抓取信息的能力和其对Spark性能的影响。具体来说，若植入的代码不足，则不能抓取到足够的Spark运行时信息；反之，则可能导致Spark运行性能下降。4.3节的实验结果说明植入代码和监控系统对Spark作业的影响在合理范围以内。

Spark的Executor维护一个线程池，其中的所有任务都执行在这个线程池中，每个任务占用一个线程。通过累加Executor中所有任务线程的CPU使用率，可以计算出该Executor此时的CPU用量的总和。

Spark任务的内存由存储内存(storage memory)和执行内存(execution memory)两部分组成。其中，存储内存是指任务的输入数据占用的内存。植入的代码通过监控该输入数据对象的大小以获取存储内存。另一方面，执行内存是指任务执行过程中，洗牌(shuffle)、聚合(aggregate)和连接(join)操作产生的中间数据结构占用的内存。Spark内部维护了一个记录当前执行内存消耗情况的数据结构。通过读取解析该数据结构，可以实时获取Spark的执行内存用量。

本监控系统中，向Spark框架内部新加入一个监控管理器模块(TracingManager)，以获取Spark内部运行时信息。监控管理器的主要功能包括向监控系统：1) 注册任务；2) 注销任务；3) 周期性报告任务CPU使用率：4) 周期性报告任务内存使用量。监控管理器周期性报告的时间间隔为τ，且可由用户指定。由于任务网络传输速率和磁盘读写速率不能从Spark内部准确地获取，本文将在2.2节中介绍从Docker中获取这两种信息的方法。在Spark中植入的主要方法及其功能如表 1所示。

2.2 获取Docker信息

Spark作业启动时的同时，会向监控系统注册，注册的信息包括作业启动时间和使用Docker的标识。本监控系统通过该标识在操作系统中定位Docker容器，并启动一个Docker监控器(DockerMonitor)。Docker监控器周期性读取Docker的API文件以获取在其中运行的Executor的网络流量和磁盘使用率。

Docker关于网络流量和磁盘读写的API文件中分别存储Docker自启动以来网络流量和磁盘读写量的总和，若要获取网络传输率和磁盘读写速率则需要进一步计算。本监控系统使用以下公式计算Docker在某一时刻t的网络传输(或接收)速率：

$Net\_Rate = \left( {Total\_Trafi{c_t} - Total\_Trafi{c_{t - \tau }}} \right)/\tau $

类似地，磁盘读取(或写入)速率用以下公式计算：

Disk_Rate=(Total_Byte_t-Total_Byte_t-τ)/τ

其中：τ为监控系统每次读取API文件的时间间隔；Total_Trafic_t和Total_Byte_t分别表示Docker自启动到时刻t，网络传输(或接收)和与磁盘读取(或写入)字节数总和。

2.3 信息整合与存储

为了获取作业阶段在某一时刻对所有资源的使用情况，本监控系统需要整合来自植入代码和Docker监控器的信息。植入代码和Docker监控器获取的信息中除包含资源用量外，还有该条信息抓取时的时间戳和阶段标识。例如，任务job_i的第j个阶段stage_j在时刻t的CPU使用量为v，这条信息可表示为三元组[job_i.stage_ j, v, t]。利用任务标识，可以唯一确定阶段从属的作业。经过整合之后，作业在时刻t对所有类型资源的使用情况表示为一个6维资源消耗向量：

M=(m₁, m₂, m₃, m₄, m₅, m₆)

其中: m₁为CPU使用率；m₂为内存用量；m₃为磁盘读取速率；m₄为磁盘写入速率；m₅为网络接收速率；m₆为网络传输速率。

为了能重复使用监控信息，需要将监控信息持久化。本监控系统使用Graphite^[30]作为后台数据库。Graphite数据库适用于存储带时间戳的数字信息，其每条数据由三元组[标识，数据的值，时间戳]组成，因此本监控系统可将整合后的信息三元组直接存储到数据库中。

2.4 模型训练

在2.2节和2.3节中抓取的原始数据包含噪声，并且资源消耗向量各维度的量纲不同，需要对数据进行预处理。在数据分析模块检测异常作业之前，还需要训练高斯混合模型。

2.4.1 数据预处理

经过察看发现，作业的每个阶段刚启动时很短一段时间内，其资源消耗向量中的每一项值都为0。这是由于作业阶段从向监控系统注册到真正启动执行需要一段准备时间。在数据预处理时，需要去除这些全零向量，以降低噪声的干扰。

由于资源消耗向量各维度的量纲不同，为了使训练的结果更准确，需要将数据标准化。本文选用离差标准化的方法，经过标准化后的样本值都被映射到[0, 1]。在资源消耗向量中，CPU使用率本身已在[0, 1]，而其他5项都需要进行标准化处理。离差标准化公式如下：

x^*=(x-μ)/(x_max-x_min)

其中：x和x^*分别为标准化前后的样本值；μ为样本方差；x_min和x_min分别为样本中的最大值和最小值。

2.4.2 参数训练

在训练参数和检测异常阶段，本文都以主机节点作为单位，也就是说——经过训练以后，每一台主机节点拥有各自独立的模型参数，并只检查在该节点上运行的作业。当Spark部署在异构集群中(heterogeneous cluster)中时，不同硬件配置的主机节点会对作业的性能和资源消耗情况造成不同的影响。因此，不能使用同一套参数来检测不同主机节点上的作业。各节点拥有各自独立的模型参数使监控系统能很好地够适应异构集群环境。

利用来自作业的历史数据，在参数训练阶段EM算法会建立K个高斯分类的参数，每个分类代表具有某种特征的作业阶段。由于EM算法对参数初值敏感，训练模型的过程中需要使用不同初始参数值进行多次训练，最后取使1.3节中式(2) 最大的训练结果作为模型参数。通过实验发现，K=4时，监控系统可以在有效监测异常作业的同时兼顾Spark作业的运行性能。

2.5 异常检测

Spark作业内部的不同阶段对资源消耗的情况不尽相同。例如，作业的第一个阶段通常会从磁盘读取数据，从而造成很大的磁盘I/O开销；而中间阶段通常会在节点间传输数据，从而造成网络I/O开销。因此，本监控系统以作业阶段作为异常检测的单位。

定义1   异常阶段是指在一段时间T内，有αT/τ条作业性能消耗向量属于每个分类的概率都不超过σ的阶段。其中：τ为抓取数据的时间间隔；α和σ均为实验测出的阈值，0 ≤ α, σ ≤ 1。

定义1中规定，只有当作业阶段中长时间出现不能分类的性能消耗向量时，才将该作业阶段划分为异常作业。这避免了系统资源短时间波动引起的误判。

监控系统会在Spark作业运行的同时将检测到的异常作业阶段反馈给用户。

3 系统实现

本监控系统由植入Spark的代码和外部守护进程两部分组成。在Spark中植入了约500行Java代码和约100行Scala代码，植入代码包括修改后的Executor和新加入的监控管理器模块(TracingManager)。修改后的Executor在任务启动和完成时向监控管理器报告；监控管理器利用Java提供的API获取Spark作业的CPU使用率和输入数据对象大小，并与外部守护进程通信。

守护进程用大约3600行Java代码编写。守护进程的主要功能包括：1) 记录正在运行的Spark作业；2) 实时获取Docker资源用量；3) 实时分析资源用量并报告异常作业；4) 将数据存储到数据库。植入代码和守护进程间使用Apache Thrift^[31]协议通信。本文选用Graphite作为后台数据库。Graphite作为企业级时间序列数据库，适用于存储带时间戳的监控数据。

4 实验结果与分析

本文针对监控系统对Spark性能的影响和异常检测的有效性设计了两组实验，并与其他监控分析工具进行对比。

在性能分析实验中，部分其他工具采用离线分析模式(即当作业完成之后，工具再对作业日志进行分析)，无法获取其对作业性能的直接影响；因此，该实验选取Whodunit^[10]、Gist^[18]和Stitch^[8]等在线监控工具进行对比。

在有效性分析实验中，部分其他工具仅仅向用户反馈带时间戳的事件流，而异常检测需要由用户完成，因此该实验选取可反馈异常信息的工具Iprof^[7]进行对比。由于Iprof是离线分析工具，本文不比较其对作业性能的影响。

4.1 实验环境

为了测试监控系统的性能，本文用9台小型服务器搭建了Spark on YARN分布式环境，1台作为Master节点，8台作为Worker节点。9台服务器的配置均为Intel Core i7-2600 @ 3.4 GHz 8核CPU、8 GB内存、500 GB/7200 rpm；服务器间用千兆网络连接。操作系统版本为Ubuntu Server 16.04。基于Spark-2.1.0版本植入代码，使用cluster模式运行在hadoop-2.7.3版本上。Docker镜像版本为sequenceiq/hadoop-docker-2.4.0。Graphite数据库版本为0.10，部署在Master节点上。

测试数据选用标准测试集HiBench-6.0^[32]的5个作业：wordcount、terasort、k-means、baye和pagerank，每个作业数据量及说明如表 2所示。

4.2 性能分析

监控系统对Spark性能的影响应在合理范围以内，以保证Spark作业正常运行。为此，本文对比了Spark单独运行和Spark与监控系统同时运行时的性能，实验结果如图 3所示。实验表明，监控系统仅使作业的执行时间增加了平均4.7%。对于非CPU密集型作业，如wordcount和terasort，监控系统对作业性能的影响不到5%。本文方法与其他监控系统对作业性能影响的对比如表 3所示。由表 3可以看出，本文方法对作业性能的影响与Whodunit和Gist大致相当，且优于Stitch。

4.3 有效性分析

为了检验监控系统对异常作业的检测的有效性，在有其他作业干扰的情况下运行HiBench Spark测试作业集，并将结果与Iprof^[7]对比。本文规定，一个作业阶段的执行时间若比无干扰时慢20%或以上则为异常作业阶段。干扰作业来自HiBench的Hadoop MapReduce测试作业集，在Spark作业运行的同时随机选取执行。

实验中，每个Spark作业在有干扰的情况下重复多次执行。首先人工检测Spark作业的异常阶段，然后与监控系统反馈的异常阶段对比，以验证监控系统的有效性。在2.5节中，本文给出了异常作业的定义。经过实验发现，定义中的阈值取α =0.6、σ =0.4时，监控系统可以较好地检测出异常作业阶段。表 4为有效性测试的实验结果。从表 4中可看出，监控系统对异常阶段检测的整体有效率为90.2%。本文监控系统与Iprof异常检测有效性分别为90.2%、73.0%。

5 结语

本文针对分布式系统性能监控及诊断困难的问题，提出并编写了一套用于Spark作业性能监控与分析的系统。该系统基于高斯混合模型，能够实时监控Spark作业的资源消耗情况，并向用户反馈性能受到干扰的作业。随后，用户可采取进一步的措施以调整Spark作业，使其恢复正常运行。实验结果表明该系统在有效检测异常作业的同时对作业性能造成的影响很小。

进一步的研究可从以下三个方向展开：1) 优化异常检测算法，使其具有更强的自动检测能力；2) 进一步分析并找出使作业性能下降的瓶颈资源；3) 利用资源消耗信息进行作业调度，最大化利用集群资源。