Architecture Phase1 Week13:Summarize

本周学习得主要是大数据中得流处理，PageRank，Knn，贝叶斯分类，K-Means 得分类算法，推荐算法，机器学习架构。

Spark

Spark 特点（Spark 为什么更快）

DAG 切分的多阶段计算过程更快速

使用内存存储中间计算结果更高效

RDD 的编程模型更简单

Spark WordCount 编程示例

val textFile = sc.textFile("hdfs://...")

val counts = textFile.flatMap(line => line.split(" "))

.map(word => (word, 1))

.reduceByKey(_ + _)

counts.saveAsTextFile("hdfs://...")

第 1 行代码：根据 HDFS 路径生成一个输入数据 RDD。

第 2 行代码：在输入数据 RDD 上执行 3 个操作，得到一个新的 RDD。

• 将输入数据的每一行文本用空格拆分成单词。

• 将每个单词进行转换，word => (word, 1)，生成 的结构。

• 相同的 Key 进行统计，统计方式是对 Value 求和，(_ + _)。

第 3 行代码：将这个 RDD 保存到 HDFS

<

作为编程模型的 RDD

RDD 是 Spark 的核心概念，是弹性分布式数据集（Resilient Distributed Datasets）的

缩写。RDD 既是 Spark 面向开发者的编程模型，又是 Spark 自身架构的核心元素。

作为 Spark 编程模型的 RDD。我们知道，大数据计算就是在大规模的数据集上进行一系

列的数据计算处理。MapReduce 针对输入数据，将计算过程分为两个阶段，一个 Map

阶段，一个 Reduce 阶段，可以理解成是面向过程的大数据计算。我们在用

MapReduce 编程的时候，思考的是，如何将计算逻辑用 Map 和 Reduce 两个阶段实现，

map 和 reduce 函数的输入和输出是什么，MapReduce 是面向过程的。

而 Spark 则直接针对数据进行编程，将大规模数据集合抽象成一个 RDD 对象，然后在

这个 RDD 上进行各种计算处理，得到一个新的 RDD，继续计算处理，直到得到最后的

结果数据。所以 Spark 可以理解成是面向对象的大数据计算。我们在进行 Spark 编程的

时候，思考的是一个 RDD 对象需要经过什么样的操作，转换成另一个 RDD 对象，思考

的重心和落脚点都在 RDD 上。

作为数据分片的 RDD

跟 MapReduce 一样，Spark 也是对大数据进行分片计算，Spark 分布式计算的数据分

片、任务调度都是以 RDD 为单位展开的，每个 RDD 分片都会分配到一个执行进程去处

理。

RDD 上的转换操作又分成两种，一种转换操作产生的 RDD 不会出现新的分片，比如

map、filter 等，也就是说一个 RDD 数据分片，经过 map 或者 filter 转换操作后，结

果还在当前分片。就像你用 map 函数对每个数据加 1，得到的还是这样一组数据，只是

值不同。实际上，Spark 并不是按照代码写的操作顺序去生成 RDD，比如

rdd2 = rdd1.map(func)

这样的代码并不会在物理上生成一个新的 RDD。物理上，Spark 只有在产生新的 RDD

分片时候，才会在物理上真的生成一个 RDD，Spark 的这种特性也被称作惰性计算。

Spark 的计算阶段

和 MapReduce 一个应用一次只运行一个 map 和一个 reduce 不同，Spark 可以根据

应用的复杂程度，分割成更多的计算阶段（stage），这些计算阶段组成一个有向无环图

DAG，Spark 任务调度器可以根据 DAG 的依赖关系执行计算阶段

这个 DAG 对应的 Spark 程序伪代码如下

rddB = rddA.groupBy(key)

rddD = rddC.map(func)

rddF = rddD.union(rddE)

rddG = rddB.join(rddF)

整个应用被切分成 3 个阶段，阶段 3 需要依赖阶段 1 和阶段 2，阶段 1 和阶段 2 互不依

赖。Spark 在执行调度的时候，先执行阶段 1 和阶段 2，完成以后，再执行阶段 3。如

果有更多的阶段，Spark 的策略也是一样的。只要根据程序初始化好 DAG，就建立了依

赖关系，然后根据依赖关系顺序执行各个计算阶段，Spark 大数据应用的计算就完成了。

Spark 作业调度执行的核心是 DAG，有了 DAG，整个应用就被切分成哪些阶段，每个阶段的依

赖关系也就清楚了。之后再根据每个阶段要处理的数据量生成相应的任务集合（TaskSet），每

个任务都分配一个任务进程去处理，Spark 就实现了大数据的分布式计算。

负责 Spark 应用 DAG 生成和管理的组件是 DAGScheduler，DAGScheduler 根据程序代码生

成 DAG，然后将程序分发到分布式计算集群，按计算阶段的先后关系调度执行。

那么 Spark 划分计算阶段的依据是什么呢？显然并不是 RDD 上的每个转换函数都会生成一个计

算阶段，比如上面的例子有 4 个转换函数，但是只有 3 个阶段。

当 RDD 之间的转换连接线呈现多对多交叉连接的时候，就会产生新的阶段。一个 RDD 代表一

个数据集，图中每个 RDD 里面都包含多个小块，每个小块代表 RDD 的一个分片。

Spark 也需要通过 shuffle 将数据进行重新组合，相同 Key 的数据放在一起，进行聚合、

关联等操作，因而每次 shuffle 都产生新的计算阶段。这也是为什么计算阶段会有依赖关

系，它需要的数据来源于前面一个或多个计算阶段产生的数据，必须等待前面的阶段执

行完毕才能进行 shuffle，并得到数据。

计算阶段划分的依据是 shuffle，不是转换函数的类型，有的函数有时候有 shuffle，有时

候没有。例子中 RDD B 和 RDD F 进行 join，得到 RDD G，这里的 RDD F 需要进行

shuffle，RDD B 就不需要。

Spark 的作业管理

Spark 里面的 RDD 函数有两种，一种是转换函数，调用以后得到的还是一个 RDD，RDD 的计算

逻辑主要通过转换函数完成。

另一种是 action 函数，调用以后不再返回 RDD。比如 count() 函数，返回 RDD 中数据的元素个

数；saveAsTextFile(path)，将 RDD 数据存储到 path 路径下。Spark 的 DAGScheduler 在遇到

shuffle 的时候，会生成一个计算阶段，在遇到 action 函数的时候，会生成一个作业（job）。

RDD 里面的每个数据分片，Spark 都会创建一个计算任务去处理，所以一个计算阶段会包含很多

个计算任务（task）。

Spark 执行过程

Spark 生态体系

流计算

Storm 实时的 Hadoop

实时计算系统

• 低延迟

• 高性能

• 分布式

• 可伸缩

• 高可用

• Nimbus：负责资源分配和任务调度。

• Supervisor：负责接受 Nimbus 分配的任务，启动和停止属于自己管理的 Worker 进程。

• Worker：运行具体处理组件逻辑的进程。

• Task：Worker 中每一个 Spout/Bolt 的线程称为一个 Task。

Topology：Storm 中运行的一个实时应用程序，因为各个组件间的消息流动形成逻辑上的一个拓扑结构。

Spout：在一个 Topology 中产生源数据流的组件。通常情况下 Spout 会从外部数据源中读取数据，然后转换

为 Topology 内部的源数据。Spout 是一个主动的角色，其接口中有个 nextTuple() 函数，Storm 框架会不停

地调用此函数，用户只要在其中生成源数据即可。

Bolt：在一个 Topology 中接受数据然后执行处理的组件。Bolt 可以执行过滤、函数操作、合并、写数据库等

任何操作。Bolt 是一个被动的角色，其接口中有个 execute(Tuple input) 函数,在接受到消息后会调用此函数，

用户可以在其中执行自己想要的操作。

Tuple：一次消息传递的基本单元。本来应该是一个 key-value 的 map，但是由于各个组件间传递的 tuple

的字段名称已经事先定义好，所以 tuple 中只要按序填入各个 value 就行了，所以就是一个 value list.

Stream：源源不断传递的 tuple 就组成了 stream

Storm 被广泛用来进行实时日志处理，出现在实时统计、实时风控、实时推荐等场景中。

一般来说，我们从类 Kafka 的 metaq 或者基于 HBase 的 TimeTunnel 中读取实时日志

消息，经过一系列处理，最终将处理结果写入到一个分布式存储中，提供给应用程序访

问。我们每天的实时消息量从几百万到几十亿不等，数据总量达到 TB 级。对于我们来说，

Storm 往往会配合分布式存储服务一起使用。在我们正在进行的个性化搜索实时分析项

目中，就使用了 TimeTunnel + HBase + Storm + UPS 的架构，每天处理几十亿的用户

日志信息，从用户行为发生到完成分析延迟在秒级。

Spark Streaming

Flink

Flink 流处理计算

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

Flink 批处理计算

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

DataSet<String> text = env.readTextFile("/path/to/file");

HiBench

HiBench 是 Intel 开放的一个 Hadoop Benchmark Suit，包含 9 个典型的 Hadoop 负载

• Micro benchmarks

Ø Sort

Ø WordCount

Ø TeraSort

• HDFS benchmarks

Ø DFSIO

• web search benchmarks

Ø Nutch indexing

Ø PageRank

• machine learning benchmarks

Ø Mahout Bayesian classification

Ø Mahout K-means clustering

• data analytics benchmarks

Ø Hive Query Benchmarks

主页是：https://github.com/intel-hadoop/hibench

Micro Benchmarks

Sort（sort）：使用 Hadoop RandomTextWriter 生成数据，并对数据进行排序。

WordCount（wordcount）：统计输入数据中每个单词的出现次数，输入数据使用

Hadoop RandomTextWriter 生成。

TeraSort（terasort）：这是由微软的数据库大牛 Jim Gray（2007 年失踪）创建的标准

benchmark，输入数据由 Hadoop TeraGen 产生。

HDFS Benchmarks

增强的 DFSIO（dfsioe）：通过产生大量同时执行读写请求的任务来测试 Hadoop 集群

的 HDFS 吞吐量

Web Search Benchmarks

Nutch indexing（nutchindexing）：大规模搜索引擎索引是 MapReduce 的一个重要应

用，这个负载测试 Nutch（Apache 的一个开源搜索引擎）的索引子系统，使用自动生成

的 Web 数据，Web 数据中的链接和单词符合 Zipfian 分布。

PageRank（pagerank）：这个负载包含一种在 Hadoop 上的 PageRank 算法实现，使

用自动生成的 Web 数据，Web 数据中的链接符合 Zipfian 分布。

Data Analytics Benchmarks

Hive Query Benchmarks（hivebench）：这个负载的开发基于 SIGMOD 09 的一篇论

文“A Comparison of Approaches to Large-Scale Data Analysis”和 HIVE-396，包含

执行典型 OLAP 查询的 Hive 查询（Aggregation and Join），使用自动生成的 Web 数

据，Web 数据中的链接符合 Zipfian 分布。

Machine Learning Benchmarks

Mahout Bayesian classification（bayes）：大规模机器学习也是 MapReduce 的一个重

要应用，这个负载测试 Mahout 0.7（Apache 的一个开源机器学习库）中的 Naive

Bayesian 训练器，输入数据是自动生成的文档，文档中的单词符合 Zipfian 分布。

Mahout K-means clustering（kmeans）：这个负载测试 Mahout 0.7 中的 K-means 聚

类算法，输入数据集由基于均匀分布和高斯分布的 GenKMeansDataset 产生

大数据可视化

比较有趣得地方，前端得展现总是比较虚幻得，可真可假。结合大数据做好随机合理推断预测得大数据更能表现出实时动态预测性。

大数据算法与机器学习

网页排名算法 PageRank

PageRank，网页排名，又称网页级别，Google 左侧排名或佩奇排名，是一种由搜索引

擎根据网页之间相互的超链接计算的技術，而作为网页排名的要素之一，以 Google 公

司創辦人拉里·佩奇（Larry Page）之姓來命名

PageRank 通过网络浩瀚的超链接關係来确定一个页面的等级。Google 把从 A 页面到

B 页面的链接解释为 A 页面给 B 页面投票，Google 根据投票来源（甚至来源的来源，

即链接到 A 页面的页面）和投票目标的等级来决定新的等级。简单的说，一个高等级的

页面可以使其他低等级页面的等级提升。

一个页面的「得票数」由所有链向它的页面的重要性來决定，到一个页面的超链接相当

于对该页投一票。一个页面的 PageRank 是由所有链向它的页面（「链入页面」）的重

要性经过递归算法得到的。一个有較多链入的页面会有較高的等级，相反如果一个页面

没有任何链入页面，那么它没有等级

KNN 分类算法

KNN 算法，也叫 K 近邻（K Nearest Neighbour）算法对于一个需要分类的数据，将其和一组已经分类标注好的样本集合进行比较，得到距离最近的 K 个样本，K 个样本最多归属的类别，就是这个需要分类数据的类别。

提取文本的特征值 TF-IDF 算法

贝叶斯分类算法

K-means 聚类算法

第 1 步：随机在图中取 K 个种子点，图中 K=2，即图中的实心小圆点。

第 2 步：求图中所有点到这 K 个种子点的距离，假如一个点离种子点 X 最近，那么这个点属于 X 点群在图中，可以看到 A、B 属于上方的种子点，C、D、E 属于中部的种子点。

第 3 步：对已经分好组的两组数据，分别求其中心点。对于图中二维平面上的数据，求中心点最简单暴力的算法就是对当前同一个分组中所有点的 X 坐标和 Y 坐标分别求平均值，得到的 就是中心点。

第 4 步：重复第 2 步和第 3 步，直到每个分组的中心点不再移动。这时候，距每个中心点最近的点数据聚类为同一组数据。

机器学习系统架构

主要得就是样本模型算法。

机器学习的数学原理

给定模型类型，也就是给定函数类型的情况下，如何寻找使结构风险最小的函数表达式。

由于函数类型已经给定，实际上就是求函数的参数。各种有样本的机器学习算法基本上

都是在各种模型的假设空间上求解结构风险最小值的过程，理解了这一点也就理解了各

种机器学习算法的推导过程。

机器学习要从假设空间寻找最优函数，而最优函数就是使样本数据的函数值和真实值距

离最小的那个函数。给定函数模型，求最优函数就是求函数的参数值。给定不同参数，

得到不同函数值和真实值的距离，这个距离就是损失，损失函数是关于模型参数的函数，

距离越小，损失越小。最小损失值对应的函数参数就是最优函数。

数学上求极小值就是求一阶导数，计算每个参数的一阶导数为零的偏微分方程组，就可

以算出最优函数的参数值。这就是为什么机器学习要计算偏微分方程的原因。

神经网络