Spark 入门

1. Spark 概述

目标

1. Spark 是什么
2. Spark 的特点
3. Spark 生态圈的组成

1.1. Spark是什么

目标

1. 了解 Spark 的历史和产生原因, 从而浅显的理解 Spark 的作用

Spark的历史

• 2009 年由加州大学伯克利分校 AMPLab 开创
• 2010 年通过BSD许可协议开源发布
• 2013 年捐赠给Apache软件基金会并切换开源协议到切换许可协议至 Apache2.0
• 2014 年 2 月，Spark 成为 Apache 的顶级项目
• 2014 年 11 月, Spark的母公司Databricks团队使用Spark刷新数据排序世界记录

Spark是什么

Apache Spark 是一个快速的, 多用途的集群计算系统, 相对于 Hadoop MapReduce 将中间结果保存在磁盘中, Spark 使用了内存保存中间结果, 能在数据尚未写入硬盘时在内存中进行运算.

Spark 只是一个计算框架, 不像 Hadoop 一样包含了分布式文件系统和完备的调度系统, 如果要使用 Spark, 需要搭载其它的文件系统和更成熟的调度系统

为什么会有Spark

Spark 产生之前, 已经有非常成熟的计算系统存在了, 例如 MapReduce, 这些计算系统提供了高层次的API, 把计算运行在集群中并提供容错能力, 从而实现分布式计算.

虽然这些框架提供了大量的对访问利用计算资源的抽象, 但是它们缺少了对利用分布式内存的抽象, 这些框架多个计算之间的数据复用就是将中间数据写到一个稳定的文件系统中(例如HDFS), 所以会产生数据的复制备份, 磁盘的I/O以及数据的序列化, 所以这些框架在遇到需要在多个计算之间复用中间结果的操作时会非常的不高效.

而这类操作是非常常见的, 例如迭代式计算, 交互式数据挖掘, 图计算等.

认识到这个问题后, 学术界的 AMPLab 提出了一个新的模型, 叫做 RDDs.

RDDs 是一个可以容错且并行的数据结构, 它可以让用户显式的将中间结果数据集保存在内中, 并且通过控制数据集的分区来达到数据存放处理最优化.

同时 RDDs 也提供了丰富的 API 来操作数据集.

后来 RDDs 被 AMPLab 在一个叫做 Spark 的框架中提供并开源.

总结

1. Spark 是Apache的开源框架
2. Spark 的母公司叫做 Databricks
3. Spark 是为了解决 MapReduce 等过去的计算系统无法在内存中保存中间结果的问题
4. Spark 的核心是 RDDs, RDDs 不仅是一种计算框架, 也是一种数据结构

1.2. Spark的特点(优点)

目标

1. 理解 Spark 的特点, 从而理解为什么要使用 Spark

速度快

• Spark 的在内存时的运行速度是 Hadoop MapReduce 的100倍
• 基于硬盘的运算速度大概是 Hadoop MapReduce 的10倍
• Spark 实现了一种叫做 RDDs 的 DAG 执行引擎, 其数据缓存在内存中可以进行迭代处理

易用

1
2
3
4

df = spark.read.json("logs.json")
df.where("age > 21") \
  .select("name.first") \
  .show()

• Spark 支持 Java, Scala, Python, R, SQL 等多种语言的API.
• Spark 支持超过80个高级运算符使得用户非常轻易的构建并行计算程序
• Spark 可以使用基于 Scala, Python, R, SQL的 Shell 交互式查询.

通用

• Spark 提供一个完整的技术栈, 包括 SQL执行, Dataset命令式API, 机器学习库MLlib, 图计算框架GraphX, 流计算SparkStreaming
• 用户可以在同一个应用中同时使用这些工具, 这一点是划时代的

兼容

• Spark 可以运行在 Hadoop Yarn, Apache Mesos, Kubernets, Spark Standalone等集群中
• Spark 可以访问 HBase, HDFS, Hive, Cassandra 在内的多种数据库

总结

• 支持 Java, Scala, Python 和 R 的 API
• 可扩展至超过 8K 个节点
• 能够在内存中缓存数据集, 以实现交互式数据分析
• 提供命令行窗口, 减少探索式的数据分析的反应时间

1.3. Spark组件

目标

1. 理解 Spark 能做什么
2. 理解 Spark 的学习路线

Spark 最核心的功能是 RDDs, RDDs 存在于 spark-core 这个包内, 这个包也是 Spark 最核心的包.

同时 Spark 在 spark-core 的上层提供了很多工具, 以便于适应不用类型的计算.

Spark-Core 和 弹性分布式数据集(RDDs)

• Spark-Core 是整个 Spark 的基础, 提供了分布式任务调度和基本的 I/O 功能
• Spark 的基础的程序抽象是弹性分布式数据集(RDDs), 是一个可以并行操作, 有容错的数据集合
  • RDDs 可以通过引用外部存储系统的数据集创建(如HDFS, HBase), 或者通过现有的 RDDs 转换得到
  • RDDs 抽象提供了 Java, Scala, Python 等语言的API
  • RDDs 简化了编程复杂性, 操作 RDDs 类似通过 Scala 或者 Java8 的 Streaming 操作本地数据集合

Spark SQL

• Spark SQL 在 spark-core 基础之上带出了一个名为 DataSet 和 DataFrame 的数据抽象化的概念
• Spark SQL 提供了在 Dataset 和 DataFrame 之上执行 SQL 的能力
• Spark SQL 提供了 DSL, 可以通过 Scala, Java, Python 等语言操作 DataSet 和 DataFrame
• 它还支持使用 JDBC/ODBC 服务器操作 SQL 语言

Spark Streaming

• Spark Streaming 充分利用 spark-core 的快速调度能力来运行流分析
• 它截取小批量的数据并可以对之运行 RDD Transformation
• 它提供了在同一个程序中同时使用流分析和批量分析的能力

MLlib

• MLlib 是 Spark 上分布式机器学习的框架. Spark分布式内存的架构 比 Hadoop磁盘式 的 Apache Mahout 快上 10 倍, 扩展性也非常优良
• MLlib 可以使用许多常见的机器学习和统计算法, 简化大规模机器学习
• 汇总统计, 相关性, 分层抽样, 假设检定, 随即数据生成
• 支持向量机, 回归, 线性回归, 逻辑回归, 决策树, 朴素贝叶斯
• 协同过滤, ALS
• K-means
• SVD奇异值分解, PCA主成分分析
• TF-IDF, Word2Vec, StandardScaler
• SGD随机梯度下降, L-BFGS

GraphX

GraphX 是分布式图计算框架, 提供了一组可以表达图计算的 API, GraphX 还对这种抽象化提供了优化运行

总结

• Spark 提供了 批处理(RDDs), 结构化查询(DataFrame), 流计算(SparkStreaming), 机器学习(MLlib), 图计算(GraphX) 等组件
• 这些组件均是依托于通用的计算引擎 RDDs 而构建出的, 所以 spark-core 的 RDDs 是整个 Spark 的基础

1.4. Spark和Hadoop的异同

2. Spark 集群搭建

目标

1. 从 Spark 的集群架构开始, 理解分布式环境, 以及 Spark 的运行原理
2. 理解 Spark 的集群搭建, 包括高可用的搭建方式

2.1. Spark 集群结构

目标

1. 通过应用运行流程, 理解分布式调度的基础概念

	Spark 如何将程序运行在一个集群中?Spark 自身是没有集群管理工具的, 但是如果想要管理数以千计台机器的集群, 没有一个集群管理工具还不太现实, 所以 Spark 可以借助外部的集群工具来进行管理整个流程就是使用 Spark 的 Client 提交任务, 找到集群管理工具申请资源, 后将计算任务分发到集群中运行

名词解释

• Driver

  该进程调用 Spark 程序的 main 方法, 并且启动 SparkContext

• Cluster Manager

  该进程负责和外部集群工具打交道, 申请或释放集群资源

• Worker

  该进程是一个守护进程, 负责启动和管理 Executor

• Executor

  该进程是一个JVM虚拟机, 负责运行 Spark Task

运行一个 Spark 程序大致经历如下几个步骤

1. 启动 Drive, 创建 SparkContext
2. Client 提交程序给 Drive, Drive 向 Cluster Manager 申请集群资源
3. 资源申请完毕, 在 Worker 中启动 Executor
4. Driver 将程序转化为 Tasks, 分发给 Executor 执行

问题一: Spark 程序可以运行在什么地方?

Spark 可以将任务运行在两种模式下:

• 单机, 使用线程模拟并行来运行程序
• 集群, 使用集群管理器来和不同类型的集群交互, 将任务运行在集群中

Spark 可以使用的集群管理工具有:

• Spark Standalone
• Hadoop Yarn
• Apache Mesos
• Kubernetes

问题二: Driver 和 Worker 什么时候被启动?

• Standalone 集群中, 分为两个角色: Master 和 Slave, 而 Slave 就是 Worker, 所以在 Standalone 集群中, 启动之初就会创建固定数量的 Worker
• Driver 的启动分为两种模式: Client 和 Cluster. 在 Client 模式下, Driver 运行在 Client 端, 在 Client 启动的时候被启动. 在 Cluster 模式下, Driver 运行在某个 Worker 中, 随着应用的提交而启动

• 在 Yarn 集群模式下, 也依然分为 Client 模式和 Cluster 模式, 较新的版本中已经逐渐在废弃 Client 模式了, 所以上图所示为 Cluster 模式
• 如果要在 Yarn 中运行 Spark 程序, 首先会和 RM 交互, 开启 ApplicationMaster, 其中运行了 Driver, Driver创建基础环境后, 会由 RM 提供对应的容器, 运行 Executor, Executor会反向向 Driver 反向注册自己, 并申请 Tasks 执行
• 在后续的 Spark 任务调度部分, 会更详细介绍

总结

• Master 负责总控, 调度, 管理和协调 Worker, 保留资源状况等
• Slave 对应 Worker 节点, 用于启动 Executor 执行 Tasks, 定期向 Master汇报
• Driver 运行在 Client 或者 Slave(Worker) 中, 默认运行在 Slave(Worker) 中

2.2. Spark 集群搭建

目标

1. 大致了解 Spark Standalone 集群搭建的过程

   这个部分的目的是搭建一套用于测试和学习的集群, 实际的工作中可能集群环境会更复杂一些

Step 1 下载和解压

1. 下载 Spark 安装包, 下载时候选择对应的 Hadoop 版本(资料中已经提供了 Spark 安装包, 直接上传至集群 Master 即可, 无需遵循以下步骤)

   https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz

   1 2 3

   # 下载 Spark cd /export/softwares wget https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz

2. 解压并拷贝到export/servers

   1 2 3 4 5

   # 解压 Spark 安装包 tar xzvf spark-2.2.0-bin-hadoop2.7.tgz # 移动 Spark 安装包 mv spark-2.2.0-bin-hadoop2.7.tgz /export/servers/spark

3. 修改配置文件spark-env.sh, 以指定运行参数

   • 进入配置目录, 并复制一份新的配置文件, 以供在此基础之上进行修改

     1 2 3

     cd /export/servers/spark/conf cp spark-env.sh.template spark-env.sh vi spark-env.sh

   • 将以下内容复制进配置文件末尾

     1 2 3 4 5 6

     # 指定 Java Home export JAVA_HOME=/export/servers/jdk1.8.0 # 指定 Spark Master 地址 export SPARK_MASTER_HOST=node01 export SPARK_MASTER_PORT=7077

Step 2 配置

1. 修改配置文件 slaves, 以指定从节点为止, 从在使用 sbin/start-all.sh 启动集群的时候, 可以一键启动整个集群所有的 Worker

   • 进入配置目录, 并复制一份新的配置文件, 以供在此基础之上进行修改

     1 2 3

     cd /export/servers/spark/conf cp slaves.template slaves vi slaves

   • 配置所有从节点的地址

     1 2	node02 node03

2. 配置 HistoryServer

   1. 默认情况下, Spark 程序运行完毕后, 就无法再查看运行记录的 Web UI 了, 通过 HistoryServer 可以提供一个服务, 通过读取日志文件, 使得我们可以在程序运行结束后, 依然能够查看运行过程

   2. 复制 spark-defaults.conf, 以供修改

      1 2 3

      cd /export/servers/spark/conf cp spark-defaults.conf.template spark-defaults.conf vi spark-defaults.conf

   3. 将以下内容复制到spark-defaults.conf末尾处, 通过这段配置, 可以指定 Spark 将日志输入到 HDFS 中

      1 2 3

      spark.eventLog.enabled true spark.eventLog.dir hdfs://node01:8020/spark_log spark.eventLog.compress true

   4. 将以下内容复制到spark-env.sh的末尾, 配置 HistoryServer 启动参数, 使得 HistoryServer 在启动的时候读取 HDFS 中写入的 Spark 日志

      1 2	# 指定 Spark History 运行参数 export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=4000 -Dspark.history.retainedApplications=3 -Dspark.history.fs.logDirectory=hdfs://node01:8020/spark_log"

   5. 为 Spark 创建 HDFS 中的日志目录

      1	hdfs dfs -mkdir -p /spark_log

Step 3 分发和运行

1. 将 Spark 安装包分发给集群中其它机器

   1 2 3

   cd /export/servers scp -r spark root@node02:$PWD scp -r spark root@node03:$PWD

2. 启动 Spark Master 和 Slaves, 以及 HistoryServer

   1 2 3

   cd /export/servers/spark sbin/start-all.sh sbin/start-history-server.sh

目标

Spark 的集群搭建大致有如下几个步骤

1. 下载和解压 Spark
2. 配置 Spark 的所有从节点位置
3. 配置 Spark History server 以便于随时查看 Spark 应用的运行历史
4. 分发和运行 Spark 集群

2.3. Spark 集群高可用搭建

目标

1. 简要了解如何使用 Zookeeper 帮助 Spark Standalone 高可用

Step 1 停止 Spark 集群

1
2

cd /export/servers/spark
sbin/stop-all.sh

Step 2 修改配置文件, 增加 Spark 运行时参数, 从而指定 Zookeeper 的位置

1. 进入 spark-env.sh 所在目录, 打开 vi 编辑

   1 2	cd /export/servers/spark/conf vi spark-env.sh

2. 编辑 spark-env.sh, 添加 Spark 启动参数, 并去掉 SPARK_MASTER_HOST 地址

   

   1 2 3 4 5 6 7 8 9 10

   # 指定 Java Home export JAVA_HOME=/export/servers/jdk1.8.0_141 # 指定 Spark Master 地址 # export SPARK_MASTER_HOST=node01 export SPARK_MASTER_PORT=7077 # 指定 Spark History 运行参数 export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=4000 -Dspark.history.retainedApplications=3 -Dspark.history.fs.logDirectory=hdfs://node01:8020/spark_log" # 指定 Spark 运行时参数 export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=node01:2181,node02:2181,node03:2181 -Dspark.deploy.zookeeper.dir=/spark"

Step 3 分发配置文件到整个集群

1
2
3

cd /export/servers/spark/conf
scp spark-env.sh node02:$PWD
scp spark-env.sh node03:$PWD

Step 4 启动

1. 在 node01 上启动整个集群

   1 2 3

   cd /export/servers/spark sbin/start-all.sh sbin/start-history-server.sh

2. 在 node02 上单独再启动一个 Master

   1 2	cd /export/servers/spark sbin/start-master.sh

Step 5 查看 node01 master 和 node02 master 的 WebUI

1. 你会发现一个是 ALIVE(主), 另外一个是 STANDBY(备)

   

2. 如果关闭一个, 则另外一个成为ALIVE, 但是这个过程可能要持续两分钟左右, 需要耐心等待

   1 2 3

   # 在 Node01 中执行如下指令 cd /export/servers/spark/ sbin/stop-master.sh

   

2.4. 第一个应用的运行

目标

1. 从示例应用运行中理解 Spark 应用的运行流程

流程

Step 1 进入 Spark 安装目录中

1

cd /export/servers/spark/

Step 2 运行 Spark 示例任务

1
2
3
4
5
6
7

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://node01:7077,node02:7077,node03:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/export/servers/spark/examples/jars/spark-examples_2.11-2.2.3.jar \
100

Step 3 运行结果

1

Pi is roughly 3.141550671141551

	刚才所运行的程序是 Spark 的一个示例程序, 使用 Spark 编写了一个以蒙特卡洛算法来计算圆周率的任务蒙特卡洛算法概述在一个正方形中, 内切出一个圆形随机向正方形内均匀投 n 个点, 其落入内切圆内的内外点的概率满足如下以上就是蒙特卡洛的大致理论, 通过这个蒙特卡洛, 便可以通过迭代循环投点的方式实现蒙特卡洛算法求圆周率

计算过程

1. 不断的生成随机的点, 根据点距离圆心是否超过半径来判断是否落入园内

2. 通过 来计算圆周率

3. 不断的迭代

4. 思考1: 迭代计算

   如果上述的程序使用 MapReduce 该如何编写? 是否会有大量的向 HDFS 写入, 后再次读取数据的做法? 是否会影响性能?

   Spark 为什么擅长这类操作? 大家可以发挥想象, 如何解决这种迭代计算的问题

5. 思考2: 数据规模

   刚才的计算只做了100次, 如果迭代100亿次, 在单机上运行和一个集群中运行谁更合适?

3. Spark 入门

目标

1. 通过理解 Spark 小案例, 来理解 Spark 应用
2. 理解编写 Spark 程序的两种常见方式
   1. spark-shell
   2. spark-submit

Spark 官方提供了两种方式编写代码, 都比较重要, 分别如下

• spark-shell
  Spark shell 是 Spark 提供的一个基于 Scala 语言的交互式解释器, 类似于 Scala 提供的交互式解释器, Spark shell 也可以直接在 Shell 中编写代码执行
  这种方式也比较重要, 因为一般的数据分析任务可能需要探索着进行, 不是一蹴而就的, 使用 Spark shell 先进行探索, 当代码稳定以后, 使用独立应用的方式来提交任务, 这样是一个比较常见的流程
• spark-submit
  Spark submit 是一个命令, 用于提交 Scala 编写的基于 Spark 框架, 这种提交方式常用作于在集群中运行任务

3.1. Spark shell 的方式编写 WordCount

概要

在初始阶段工作可以全部使用 Spark shell 完成, 它可以加快原型开发, 使得迭代更快, 很快就能看到想法的结果. 但是随着项目规模越来越大, 这种方式不利于代码维护, 所以可以编写独立应用. 一般情况下, 在探索阶段使用 Spark shell, 在最终使用独立应用的方式编写代码并使用 Maven 打包上线运行

接下来使用 Spark shell 的方式编写一个 WordCount

Step 1 准备文件

在 Node01 中创建文件 /export/data/wordcount.txt

1
2
3

hadoop spark flume
spark hadoop
flume hadoop

Step 2 启动 Spark shell

1
2

cd /export/servers/spark
bin/spark-shell --master local[2]

Step 3 执行如下代码

1
2
3
4
5
6
7
8
9

scala> val sourceRdd = sc.textFile("file:///export/data/wordcount.txt")
sourceRdd: org.apache.spark.rdd.RDD[String] = file:///export/data/wordcount.txt MapPartitionsRDD[1] at textFile at <console>:24

scala> val flattenCountRdd = sourceRdd.flatMap(.split(" ")).map((, 1))
flattenCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:26
scala> val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
aggCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:28
scala> val result = aggCountRdd.collect
result: Array[(String, Int)] = Array((spark,2), (hadoop,3), (flume,2))

运行流程

1. flatMap(_.split(" ")) 将数据转为数组的形式, 并展平为多个数据
2. map_, 1 将数据转换为元组的形式
3. reduceByKey(_ + _) 计算每个 Key 出现的次数

总结

1. 使用 Spark shell 可以快速验证想法
2. Spark 框架下的代码非常类似 Scala 的函数式调用

3.2. 读取 HDFS 上的文件

目标

1. 理解 Spark 访问 HDFS 的两种方式

Step 1 上传文件到 HDFS 中

1
2
3

cd /export/data
hdfs dfs -mkdir /dataset
hdfs dfs -put wordcount.txt /dataset/

Step 2 在 Spark shell 中访问 HDFS

1
2
3
4
5
6
7
8
9
10

val sourceRdd = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")
val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_, 1))
val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
val result = aggCountRdd.collect


val sourceRdd = sc.textFile("file///export/servers/wordcount.txt")
val flattenCountRdd = sourceRdd.flatMap(_.split(",")).map((_, 1))
val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
val result = aggCountRdd.collect

	如何使得 Spark 可以访问 HDFS?可以通过指定 HDFS 的 NameNode 地址直接访问, 类似于上面代码中的 sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")也可以通过向 Spark 配置 Hadoop 的路径, 来通过路径直接访问1.在 spark-env.sh 中添加 Hadoop 的配置路径export HADOOP_CONF_DIR="/etc/hadoop/conf"2.在配置过后, 可以直接使用 hdfs:///路径 的形式直接访问3.在配置过后, 也可以直接使用路径访问

3.4. 编写独立应用提交 Spark 任务

目标

1. 理解如何编写 Spark 独立应用
2. 理解 WordCount 的代码流程

Step 1 创建工程

1. 创建 IDEA 工程
   1. → →
   2. → →
2. 增加 Scala 支持
   1. 右键点击工程目录
   2. 选择增加框架支持
   3. 选择 Scala 添加框架支持

Step 2 编写 Maven 配置文件 pom.xml

1. 工程根目录下增加文件 pom.xml

2. 添加以下内容

3. 因为在 pom.xml 中指定了 Scala 的代码目录, 所以创建目录 src/main/scala 和目录 src/test/scala

4. 创建 Scala object WordCount

Step 3 编写代码

1

object WordCounts {   def main(args: Array[String]): Unit = {    // 1. 创建 Spark Context    val conf = new SparkConf().setMaster("local[2]")    val sc: SparkContext = new SparkContext(conf) <span class="hljs-comment">// 2. 读取文件并计算词频</span> val source: RDD[String] = sc.textFile(<span class="hljs-string">"hdfs://node01:8020/dataset/wordcount.txt"</span>, <span class="hljs-number">2</span>) val words: RDD[String] = source.flatMap { line => line.split(<span class="hljs-string">" "</span>) } val wordsTuple: RDD[(String, Int)] = words.map { word => (word, <span class="hljs-number">1</span>) } val wordsCount: RDD[(String, Int)] = wordsTuple.reduceByKey { (x, y) => x + y } <span class="hljs-comment">// 3. 查看执行结果</span> println(wordsCount.collect)  } }

Step 4 运行

运行 Spark 独立应用大致有两种方式, 一种是直接在 IDEA 中调试, 另一种是可以在提交至 Spark 集群中运行, 而 Spark 又支持多种集群, 不同的集群有不同的运行方式

直接在 IDEA 中运行 Spark 程序

1. 在工程根目录创建文件夹和文件

   

2. 修改读取文件的路径为dataset/wordcount.txt

   

3. 右键运行 Main 方法

   

提交到 Spark Standalone 集群中运行

spark-submit 命令

1

spark-submit [options] <app jar> <app options>

• app jar 程序 Jar 包
• app options 程序 Main 方法传入的参数
• options 提交应用的参数, 可以有如下选项

1. 在 IDEA 中使用 Maven 打包

   

2. 拷贝打包的 Jar 包上传到 node01 中

   

3. 在 node01 中 Jar 包所在的目录执行如下命令

   1 2 3

   spark-submit --master spark://node01:7077 \ --class cn.itcast.spark.WordCounts \ original-spark-0.1.0.jar

总结: 三种不同的运行方式

Spark shell

• 作用
  • 一般用作于探索阶段, 通过 Spark shell 快速的探索数据规律
  • 当探索阶段结束后, 代码确定以后, 通过独立应用的形式上线运行
• 功能
  • Spark shell 可以选择在集群模式下运行, 还是在线程模式下运行
  • Spark shell 是一个交互式的运行环境, 已经内置好了 SparkContext 和 SparkSession 对象, 可以直接使用
  • Spark shell 一般运行在集群中安装有 Spark client 的服务器中, 所以可以自有的访问 HDFS

本地运行

• 作用
  • 在编写独立应用的时候, 每次都要提交到集群中还是不方便, 另外很多时候需要调试程序, 所以在 IDEA 中直接运行会比较方便, 无需打包上传了
• 功能
  • 因为本地运行一般是在开发者的机器中运行, 而不是集群中, 所以很难直接使用 HDFS 等集群服务, 需要做一些本地配置, 用的比较少
  • 需要手动创建 SparkContext

集群运行

• 作用
  • 正式环境下比较多见, 独立应用编写好以后, 打包上传到集群中, 使用spark-submit来运行, 可以完整的使用集群资源
• 功能
  • 同时在集群中通过spark-submit来运行程序也可以选择是用线程模式还是集群模式
  • 集群中运行是全功能的, HDFS 的访问, Hive 的访问都比较方便
  • 需要手动创建 SparkContext

4. RDD 入门

目标

上面通过一个 WordCount 案例, 演示了 Spark 大致的编程模型和运行方式, 接下来针对 Spark 的编程模型做更详细的扩展

1. 理解 WordCount 的代码
   1. 从执行角度上理解, 数据之间如何流转
   2. 从原理角度理解, 各个算子之间如何配合
2. 粗略理解 Spark 中的编程模型 RDD
3. 理解 Spark 中 RDD 的各个算子

1

object WordCounts {   def main(args: Array[String]): Unit = {    // 1. 创建 Spark Context    val conf = new SparkConf().setMaster("local[2]")    val sc: SparkContext = new SparkContext(conf) <span class="hljs-comment">// 2. 读取文件并计算词频</span> val source: RDD[String] = sc.textFile(<span class="hljs-string">"hdfs://node01:8020/dataset/wordcount.txt"</span>, <span class="hljs-number">2</span>) val words: RDD[String] = source.flatMap { line => line.split(<span class="hljs-string">" "</span>) } val wordsTuple: RDD[(String, Int)] = words.map { word => (word, <span class="hljs-number">1</span>) } val wordsCount: RDD[(String, Int)] = wordsTuple.reduceByKey { (x, y) => x + y } <span class="hljs-comment">// 3. 查看执行结果</span> println(wordsCount.collect)  } }

在这份 WordCount 代码中, 大致的思路如下:

1. 使用 sc.textFile() 方法读取 HDFS 中的文件, 并生成一个 RDD
2. 使用 flatMap 算子将读取到的每一行字符串打散成单词, 并把每个单词变成新的行
3. 使用 map 算子将每个单词转换成 (word, 1) 这种元组形式
4. 使用 reduceByKey 统计单词对应的频率

其中所使用到的算子有如下几个:

• flatMap 是一对多
• map 是一对一
• reduceByKey 是按照 Key 聚合, 类似 MapReduce 中的 Shuffled

如果用图形表示的话, 如下:

总结以及引出新问题

上面大概说了两件事:

1. 代码流程
2. 算子

在代码中有一些东西并未交代:

1. source, words, wordsTuple 这些变量的类型是 RDD[Type], 什么是 RDD?
2. 还有更多算子吗?

RDD 是什么

定义

RDD, 全称为 Resilient Distributed Datasets, 是一个容错的, 并行的数据结构, 可以让用户显式地将数据存储到磁盘和内存中, 并能控制数据的分区.

同时, RDD 还提供了一组丰富的操作来操作这些数据. 在这些操作中, 诸如 map, flatMap, filter 等转换操作实现了 Monad 模式, 很好地契合了 Scala 的集合操作. 除此之外, RDD 还提供了诸如 join, groupBy, reduceByKey 等更为方便的操作, 以支持常见的数据运算.

通常来讲, 针对数据处理有几种常见模型, 包括: Iterative Algorithms, Relational Queries, MapReduce, Stream Processing. 例如 Hadoop MapReduce 采用了 MapReduce 模型, Storm 则采用了 Stream Processing 模型. RDD 混合了这四种模型, 使得 Spark 可以应用于各种大数据处理场景.

RDD 作为数据结构, 本质上是一个只读的分区记录集合. 一个 RDD 可以包含多个分区, 每个分区就是一个 DataSet 片段.

RDD 之间可以相互依赖, 如果 RDD 的每个分区最多只能被一个子 RDD 的一个分区使用，则称之为窄依赖, 若被多个子 RDD 的分区依赖，则称之为宽依赖. 不同的操作依据其特性, 可能会产生不同的依赖. 例如 map 操作会产生窄依赖, 而 join 操作则产生宽依赖.

特点

1. RDD 是一个编程模型
   1. RDD 允许用户显式的指定数据存放在内存或者磁盘
   2. RDD 是分布式的, 用户可以控制 RDD 的分区
2. RDD 是一个编程模型
   1. RDD 提供了丰富的操作
   2. RDD 提供了 map, flatMap, filter 等操作符, 用以实现 Monad 模式
   3. RDD 提供了 reduceByKey, groupByKey 等操作符, 用以操作 Key-Value 型数据
   4. RDD 提供了 max, min, mean 等操作符, 用以操作数字型的数据
3. RDD 是混合型的编程模型, 可以支持迭代计算, 关系查询, MapReduce, 流计算
4. RDD 是只读的
5. RDD 之间有依赖关系, 根据执行操作的操作符的不同, 依赖关系可以分为宽依赖和窄依赖

RDD 的分区

整个 WordCount 案例的程序从结构上可以用上图表示, 分为两个大部分 存储

文件如果存放在 HDFS 上, 是分块的, 类似上图所示, 这个 wordcount.txt 分了三块

计算

Spark 不止可以读取 HDFS, Spark 还可以读取很多其它的数据集, Spark 可以从数据集中创建出 RDD

例如上图中, 使用了一个 RDD 表示 HDFS 上的某一个文件, 这个文件在 HDFS 中是分三块, 那么 RDD 在读取的时候就也有三个分区, 每个 RDD 的分区对应了一个 HDFS 的分块

后续 RDD 在计算的时候, 可以更改分区, 也可以保持三个分区, 每个分区之间有依赖关系, 例如说 RDD2 的分区一依赖了 RDD1 的分区一

RDD 之所以要设计为有分区的, 是因为要进行分布式计算, 每个不同的分区可以在不同的线程, 或者进程, 甚至节点中, 从而做到并行计算

总结

1. RDD 是弹性分布式数据集
2. RDD 一个非常重要的前提和基础是 RDD 运行在分布式环境下, 其可以分区

4.1. 创建 RDD

程序入口 SparkContext

1
2

val conf = new SparkConf().setMaster("local[2]")
val sc: SparkContext = new SparkContext(conf)

SparkContext 是 spark-core 的入口组件, 是一个 Spark 程序的入口, 在 Spark 0.x 版本就已经存在 SparkContext 了, 是一个元老级的 API

如果把一个 Spark 程序分为前后端, 那么服务端就是可以运行 Spark 程序的集群, 而 Driver 就是 Spark 的前端, 在 Driver 中 SparkContext 是最主要的组件, 也是 Driver 在运行时首先会创建的组件, 是 Driver 的核心

SparkContext 从提供的 API 来看, 主要作用是连接集群, 创建 RDD, 累加器, 广播变量等

简略的说, RDD 有三种创建方式

• RDD 可以通过本地集合直接创建
• RDD 也可以通过读取外部数据集来创建
• RDD 也可以通过其它的 RDD 衍生而来

通过本地集合直接创建 RDD

1
2
3
4
5
6

val conf = new SparkConf().setMaster("local[2]")
val sc = new SparkContext(conf)

val list = List(1, 2, 3, 4, 5, 6)
val rddParallelize = sc.parallelize(list, 2)
val rddMake = sc.makeRDD(list, 2)

通过 parallelize 和 makeRDD 这两个 API 可以通过本地集合创建 RDD

这两个 API 本质上是一样的, 在 makeRDD 这个方法的内部, 最终也是调用了 parallelize

因为不是从外部直接读取数据集的, 所以没有外部的分区可以借鉴, 于是在这两个方法都都有两个参数, 第一个参数是本地集合, 第二个参数是分区数

通过读取外部文件创建 RDD

1
2
3
4

val conf = new SparkConf().setMaster("local[2]")
val sc = new SparkContext(conf)

val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")

• 访问方式
  • 支持访问文件夹, 例如 sc.textFile("hdfs:///dataset")
  • 支持访问压缩文件, 例如 sc.textFile("hdfs:///dataset/words.gz")
  • 支持通过通配符访问, 例如 sc.textFile("hdfs:///dataset/*.txt")

• 分区
  • 默认情况下读取 HDFS 中文件的时候, 每个 HDFS 的 block 对应一个 RDD 的 partition, block 的默认是128M
  • 通过第二个参数, 可以指定分区数量, 例如 sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 20)
  • 如果通过第二个参数指定了分区, 这个分区数量一定不能小于block数

• 支持的平台
  • 支持 Hadoop 的几乎所有数据格式, 支持 HDFS 的访问
  • 通过第三方的支持, 可以访问AWS和阿里云中的文件, 详情查看对应平台的 API

通过其它的 RDD 衍生新的 RDD

1
2
3
4
5

val conf = new SparkConf().setMaster("local[2]")
val sc = new SparkContext(conf)

val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 20)
val words = source.flatMap { line => line.split(" ") }

• source 是通过读取 HDFS 中的文件所创建的
• words 是通过 source 调用算子 map 生成的新 RDD

总结

1. RDD 的可以通过三种方式创建, 通过本地集合创建, 通过外部数据集创建, 通过其它的 RDD 衍生

4.2. RDD 算子

目标

1. 理解各个算子的作用
2. 通过理解算子的作用, 反向理解 WordCount 程序, 以及 Spark 的要点

Map 算子

1
2
3

sc.parallelize(Seq(1, 2, 3))
  .map( num => num * 10 )
  .collect()

作用

把 RDD 中的数据 一对一 的转为另一种形式

调用

1

def map[U: ClassTag](f: T ⇒ U): RDD[U]

参数

f → Map 算子是 原RDD → 新RDD 的过程, 这个函数的参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据

注意点

Map 是一对一, 如果函数是 String → Array[String] 则新的 RDD 中每条数据就是一个数组

FlatMap 算子

1
2
3

sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim"))
  .flatMap( line => line.split(" ") )
  .collect()

作用

FlatMap 算子和 Map 算子类似, 但是 FlatMap 是一对多

调用

1

def flatMap[U: ClassTag](f: T ⇒ List[U]): RDD[U]

参数

f → 参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据, 需要注意的是返回值是一个集合, 集合中的数据会被展平后再放入新的 RDD

注意点

flatMap 其实是两个操作, 是 map + flatten, 也就是先转换, 后把转换而来的 List 展开

ReduceByKey 算子

1
2
3

sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
  .reduceByKey( (curr, agg) => curr + agg )
  .collect()

作用

首先按照 Key 分组, 接下来把整组的 Value 计算出一个聚合值, 这个操作非常类似于 MapReduce 中的 Reduce

调用

1

def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]

参数

func → 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果

注意点

• ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2
• ReduceByKey 是一个需要 Shuffled 的操作
• 和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少

总结

1. map 和 flatMap 算子都是转换, 只是 flatMap 在转换过后会再执行展开, 所以 map 是一对一, flatMap 是一对多
2. reduceByKey 类似 MapReduce 中的 Reduce

目标

1. 深入理解 RDD 的内在逻辑
2. 能够使用 RDD 的算子
3. 理解 RDD 算子的 Shuffle 和缓存
4. 理解 RDD 整体的使用流程
5. 理解 RDD 的调度原理
6. 理解 Spark 中常见的分布式变量共享方式

深入 RDD

1. 深入 RDD

目标

1. 深入理解 RDD 的内在逻辑, 以及 RDD 的内部属性(RDD 由什么组成)

1.1. 案例

需求

• 给定一个网站的访问记录, 俗称 Access log
• 计算其中出现的独立 IP, 以及其访问的次数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)

val result = sc.textFile("dataset/access_log_sample.txt")
  .map(item => (item.split(" ")(0), 1))
  .filter(item => StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) => curr + agg)
  .sortBy(item => item._2, false)
  .take(10)

result.foreach(item => println(item))`</pre>
</div>
</div>
<div class="paragraph">

针对这个小案例, 我们问出互相关联但是又方向不同的五个问题

</div>
<div class="qlist qanda">

1. 假设要针对整个网站的历史数据进行处理, 量有 1T, 如何处理?

   放在集群中, 利用集群多台计算机来并行处理

2. 如何放在集群中运行?

   

   简单来讲, 并行计算就是同时使用多个计算资源解决一个问题, 有如下四个要点

   1	* 要解决的问题必须可以分解为多个可以并发计算的部分

   • 每个部分要可以在不同处理器上被同时执行
   • 需要一个共享内存的机制
   • 需要一个总体上的协作机制来进行调度

3. 如果放在集群中的话, 可能要对整个计算任务进行分解, 如何分解?

   

   概述

   1	* 对于 HDFS 中的文件, 是分为不同的 Block 的

   • 在进行计算的时候, 就可以按照 Block 来划分, 每一个 Block 对应一个不同的计算单元

   扩展

   1	* `RDD` 并没有真实的存放数据, 数据是从 HDFS 中读取的, 在计算的过程中读取即可

4. RDD 至少是需要可以 分片 的, 因为 HDFS 中的文件就是分片的, RDD 分片的意义在于表示对源数据集每个分片的计算, RDD 可以分片也意味着 可以并行计算

移动数据不如移动计算是一个基础的优化, 如何做到?

每一个计算单元需要记录其存储单元的位置, 尽量调度过去

在集群中运行, 需要很多节点之间配合, 出错的概率也更高, 出错了怎么办?

RDD1 → RDD2 → RDD3 这个过程中, RDD2 出错了, 有两种办法可以解决

1

1.  缓存 RDD2 的数据, 直接恢复 RDD2, 类似 HDFS 的备份机制

1. 记录 RDD2 的依赖关系, 通过其父级的 RDD 来恢复 RDD2, 这种方式会少很多数据的交互和保存

如何通过父级 RDD 来恢复?

1

1.  记录 RDD2 的父亲是 RDD1

记录 RDD2 的计算函数, 例如记录 RDD2 = RDD1.map(…), map(…) 就是计算函数

当 RDD2 计算出错的时候, 可以通过父级 RDD 和计算函数来恢复 RDD2

假如任务特别复杂, 流程特别长, 有很多 RDD 之间有依赖关系, 如何优化?

上面提到了可以使用依赖关系来进行容错, 但是如果依赖关系特别长的时候, 这种方式其实也比较低效, 这个时候就应该使用另外一种方式, 也就是记录数据集的状态

在 Spark 中有两个手段可以做到

1

1.  缓存

1. Checkpoint

1.2. 再谈 RDD

目标

理解 RDD 为什么会出现

理解 RDD 的主要特点

理解 RDD 的五大属性

1.2.1. RDD 为什么会出现?

在 RDD 出现之前, 当时 MapReduce 是比较主流的, 而 MapReduce 如何执行迭代计算的任务呢?

多个 MapReduce 任务之间没有基于内存的数据共享方式, 只能通过磁盘来进行共享

这种方式明显比较低效

RDD 如何解决迭代计算非常低效的问题呢?

在 Spark 中, 其实最终 Job3 从逻辑上的计算过程是: Job3 = (Job1.map).filter, 整个过程是共享内存的, 而不需要将中间结果存放在可靠的分布式文件系统中

这种方式可以在保证容错的前提下, 提供更多的灵活, 更快的执行速度, RDD 在执行迭代型任务时候的表现可以通过下面代码体现

1
2
3
4
5
6
7
8
9
10

`// 线性回归
val points = sc.textFile(...)
    .map(...)
    .persist(...)
val w = randomValue
for (i <- 1 to 10000) {
    val gradient = points.map(p => p.x * (1 / (1 + exp(-p.y * (w dot p.x))) - 1) * p.y)
        .reduce(_ + _)
    w -= gradient
}`

在这个例子中, 进行了大致 10000 次迭代, 如果在 MapReduce 中实现, 可能需要运行很多 Job, 每个 Job 之间都要通过 HDFS 共享结果, 熟快熟慢一窥便知

1.2.2. RDD 的特点

RDD 不仅是数据集, 也是编程模型

RDD 即是一种数据结构, 同时也提供了上层 API, 同时 RDD 的 API 和 Scala 中对集合运算的 API 非常类似, 同样也都是各种算子

RDD 的算子大致分为两类:

• Transformation 转换操作, 例如 map flatMap filter 等
• Action 动作操作, 例如 reduce collect show 等

执行 RDD 的时候, 在执行到转换操作的时候, 并不会立刻执行, 直到遇见了 Action 操作, 才会触发真正的执行, 这个特点叫做 惰性求值

RDD 可以分区

RDD 是一个分布式计算框架, 所以, 一定是要能够进行分区计算的, 只有分区了, 才能利用集群的并行计算能力

同时, RDD 不需要始终被具体化, 也就是说: RDD 中可以没有数据, 只要有足够的信息知道自己是从谁计算得来的就可以, 这是一种非常高效的容错方式

RDD 是只读的

RDD 是只读的, 不允许任何形式的修改. 虽说不能因为 RDD 和 HDFS 是只读的, 就认为分布式存储系统必须设计为只读的. 但是设计为只读的, 会显著降低问题的复杂度, 因为 RDD 需要可以容错, 可以惰性求值, 可以移动计算, 所以很难支持修改.

RDD2 中可能没有数据, 只是保留了依赖关系和计算函数, 那修改啥?

如果因为支持修改, 而必须保存数据的话, 怎么容错?

如果允许修改, 如何定位要修改的那一行? RDD 的转换是粗粒度的, 也就是说, RDD 并不感知具体每一行在哪.

RDD 是可以容错的

RDD 的容错有两种方式

保存 RDD 之间的依赖关系, 以及计算函数, 出现错误重新计算

直接将 RDD 的数据存放在外部存储系统, 出现错误直接读取, Checkpoint

1.2.3. 什么叫做弹性分布式数据集

分布式

RDD 支持分区, 可以运行在集群中

弹性

RDD 支持高效的容错

RDD 中的数据即可以缓存在内存中, 也可以缓存在磁盘中, 也可以缓存在外部存储中

数据集

RDD 可以不保存具体数据, 只保留创建自己的必备信息, 例如依赖和计算函数

RDD 也可以缓存起来, 相当于存储具体数据

总结: RDD 的五大属性

首先整理一下上面所提到的 RDD 所要实现的功能:

1. RDD 有分区
2. RDD 要可以通过依赖关系和计算函数进行容错
3. RDD 要针对数据本地性进行优化
4. RDD 支持 MapReduce 形式的计算, 所以要能够对数据进行 Shuffled

对于 RDD 来说, 其中应该有什么内容呢? 如果站在 RDD 设计者的角度上, 这个类中, 至少需要什么属性?

• Partition List 分片列表, 记录 RDD 的分片, 可以在创建 RDD 的时候指定分区数目, 也可以通过算子来生成新的 RDD 从而改变分区数目
• Compute Function 为了实现容错, 需要记录 RDD 之间转换所执行的计算函数
• RDD Dependencies RDD 之间的依赖关系, 要在 RDD 中记录其上级 RDD 是谁, 从而实现容错和计算
• Partitioner 为了执行 Shuffled 操作, 必须要有一个函数用来计算数据应该发往哪个分区
• Preferred Location 优先位置, 为了实现数据本地性操作, 从而移动计算而不是移动存储, 需要记录每个 RDD 分区最好应该放置在什么位置

2. RDD 的算子

目标

1. 理解 RDD 的算子分类, 以及其特性
2. 理解常见算子的使用

分类

RDD 中的算子从功能上分为两大类

Transformation(转换) 它会在一个已经存在的 RDD 上创建一个新的 RDD, 将旧的 RDD 的数据转换为另外一种形式后放入新的 RDD

Action(动作) 执行各个分区的计算任务, 将的到的结果返回到 Driver 中

RDD 中可以存放各种类型的数据, 那么对于不同类型的数据, RDD 又可以分为三类

• 针对基础类型(例如 String)处理的普通算子
• 针对 Key-Value 数据处理的 byKey 算子
• 针对数字类型数据处理的计算算子

特点

Spark 中所有的 Transformations 是 Lazy(惰性) 的, 它们不会立即执行获得结果. 相反, 它们只会记录在数据集上要应用的操作. 只有当需要返回结果给 Driver 时, 才会执行这些操作, 通过 DAGScheduler 和 TaskScheduler 分发到集群中运行, 这个特性叫做 惰性求值

默认情况下, 每一个 Action 运行的时候, 其所关联的所有 Transformation RDD 都会重新计算, 但是也可以使用 presist 方法将 RDD 持久化到磁盘或者内存中. 这个时候为了下次可以更快的访问, 会把数据保存到集群上.

2.1. Transformations 算子

Transformation function	解释
map(T ⇒ U)	sc.parallelize(Seq(1, 2, 3)) .map( num => num * 10 ) .collect()作用把 RDD 中的数据 一对一 的转为另一种形式签名def map[U: ClassTag](f: T ⇒ U): RDD[U]参数f → Map 算子是 原RDD → 新RDD 的过程, 传入函数的参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据注意点Map 是一对一, 如果函数是 String → Array[String] 则新的 RDD 中每条数据就是一个数组
flatMap(T ⇒ List[U])	sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim")) .flatMap( line => line.split(" ") ) .collect()作用FlatMap 算子和 Map 算子类似, 但是 FlatMap 是一对多调用def flatMap[U: ClassTag](f: T ⇒ List[U]): RDD[U]参数f → 参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据, 需要注意的是返回值是一个集合, 集合中的数据会被展平后再放入新的 RDD注意点flatMap 其实是两个操作, 是 map + flatten, 也就是先转换, 后把转换而来的 List 展开Spark 中并没有直接展平 RDD 中数组的算子, 可以使用 flatMap 做这件事
filter(T ⇒ Boolean)	sc.parallelize(Seq(1, 2, 3)) .filter( value => value >= 3 ) .collect()作用Filter 算子的主要作用是过滤掉不需要的内容
mapPartitions(List[T] ⇒ List[U])	RDD[T] ⇒ RDD[U] 和 map 类似, 但是针对整个分区的数据转换
mapPartitionsWithIndex	和 mapPartitions 类似, 只是在函数中增加了分区的 Index
mapValues	sc.parallelize(Seq(("a", 1), ("b", 2), ("c", 3))) .mapValues( value => value * 10 ) .collect())作用MapValues 只能作用于 Key-Value 型数据, 和 Map 类似, 也是使用函数按照转换数据, 不同点是 MapValues 只转换 Key-Value 中的 Value
sample(withReplacement, fraction, seed)	sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)) .sample(withReplacement = true, 0.6, 2) .collect()作用Sample 算子可以从一个数据集中抽样出来一部分, 常用作于减小数据集以保证运行速度, 并且尽可能少规律的损失参数Sample 接受第一个参数为 withReplacement, 意为是否取样以后是否还放回原数据集供下次使用, 简单的说, 如果这个参数的值为 true, 则抽样出来的数据集中可能会有重复Sample 接受第二个参数为 fraction, 意为抽样的比例Sample 接受第三个参数为 seed, 随机数种子, 用于 Sample 内部随机生成下标, 一般不指定, 使用默认值
union(other)	val rdd1 = sc.parallelize(Seq(1, 2, 3)) val rdd2 = sc.parallelize(Seq(4, 5, 6)) rdd1.union(rdd2) .collect()
intersection(other)	val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5)) val rdd2 = sc.parallelize(Seq(4, 5, 6, 7, 8)) rdd1.intersection(rdd2) .collect()作用Intersection 算子是一个集合操作, 用于求得 左侧集合 和 右侧集合 的交集, 换句话说, 就是左侧集合和右侧集合都有的元素, 并生成一个新的 RDD
subtract(other, numPartitions)	(RDD[T], RDD[T]) ⇒ RDD[T] 差集, 可以设置分区数
distinct(numPartitions)	sc.parallelize(Seq(1, 1, 2, 2, 3)) .distinct() .collect()作用Distinct 算子用于去重注意点Distinct 是一个需要 Shuffled 的操作本质上 Distinct 就是一个 reductByKey, 把重复的合并为一个
reduceByKey((V, V) ⇒ V, numPartition)	sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))) .reduceByKey( (curr, agg) => curr + agg ) .collect()作用首先按照 Key 分组生成一个 Tuple, 然后针对每个组执行 reduce 算子调用def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]参数func → 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果注意点ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2ReduceByKey 是一个需要 Shuffled 的操作和其它的 Shuffled 相比, ReduceByKey 是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少
groupByKey()	sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))) .groupByKey() .collect()作用GroupByKey 算子的主要作用是按照 Key 分组, 和 ReduceByKey 有点类似, 但是 GroupByKey 并不求聚合, 只是列举 Key 对应的所有 Value注意点GroupByKey 是一个 ShuffledGroupByKey 和 ReduceByKey 不同, 因为需要列举 Key 对应的所有数据, 所以无法在 Map 端做 Combine, 所以 GroupByKey 的性能并没有 ReduceByKey 好
combineByKey()	val rdd = sc.parallelize(Seq( ("zhangsan", 99.0), ("zhangsan", 96.0), ("lisi", 97.0), ("lisi", 98.0), ("zhangsan", 97.0)) ) val combineRdd = rdd.combineByKey( score => (score, 1), (scoreCount: (Double, Int),newScore) => (scoreCount._1 + newScore, scoreCount._2 + 1), (scoreCount1: (Double, Int), scoreCount2: (Double, Int)) => (scoreCount1._1 + scoreCount2._1, scoreCount1._2 + scoreCount2._2) ) val meanRdd = combineRdd.map(score => (score._1, score._2._1 / score._2._2)) meanRdd.collect()作用对数据集按照 Key 进行聚合调用combineByKey(createCombiner, mergeValue, mergeCombiners, [partitioner], [mapSideCombiner], [serializer])参数createCombiner 将 Value 进行初步转换mergeValue 在每个分区把上一步转换的结果聚合mergeCombiners 在所有分区上把每个分区的聚合结果聚合partitioner 可选, 分区函数mapSideCombiner 可选, 是否在 Map 端 Combineserializer 序列化器注意点combineByKey 的要点就是三个函数的意义要理解groupByKey, reduceByKey 的底层都是 combineByKey
aggregateByKey()	val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0))) val result = rdd.aggregateByKey(0.8)( seqOp = (zero, price) => price * zero, combOp = (curr, agg) => curr + agg ).collect() println(result)作用聚合所有 Key 相同的 Value, 换句话说, 按照 Key 聚合 Value调用rdd.aggregateByKey(zeroValue)(seqOp, combOp)参数zeroValue 初始值seqOp 转换每一个值的函数comboOp 将转换过的值聚合的函数注意点为什么需要两个函数? * aggregateByKey 运行将一个 RDD[(K, V)] 聚合为 RDD[(K, U)], 如果要做到这件事的话, 就需要先对数据做一次转换, 将每条数据从 V 转为 U, seqOp 就是干这件事的 ** 当 seqOp 的事情结束以后, comboOp 把其结果聚合和 reduceByKey 的区别::` aggregateByKey 最终聚合结果的类型和传入的初始值类型保持一致`reduceByKey 在集合中选取第一个值作为初始值, 并且聚合过的数据类型不能改变
foldByKey(zeroValue)((V, V) ⇒ V)	sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))) .foldByKey(zeroValue = 10)( (curr, agg) => curr + agg ) .collect()作用和 ReduceByKey 是一样的, 都是按照 Key 做分组去求聚合, 但是 FoldByKey 的不同点在于可以指定初始值调用foldByKey(zeroValue)(func)参数zeroValue 初始值func seqOp 和 combOp 相同, 都是这个参数注意点FoldByKey 是 AggregateByKey 的简化版本, seqOp 和 combOp 是同一个函数FoldByKey 指定的初始值作用于每一个 Value
join(other, numPartitions)	val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1))) val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12))) rdd1.join(rdd2).collect()作用将两个 RDD 按照相同的 Key 进行连接调用join(other, [partitioner or numPartitions])参数other 其它 RDDpartitioner or numPartitions 可选, 可以通过传递分区函数或者分区数量来改变分区注意点Join 有点类似于 SQL 中的内连接, 只会再结果中包含能够连接到的 KeyJoin 的结果是一个笛卡尔积形式, 例如 "a", 1), ("a", 2 和 "a", 10), ("a", 11 的 Join 结果集是 "a", 1, 10), ("a", 1, 11), ("a", 2, 10), ("a", 2, 11
cogroup(other, numPartitions)	val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("a", 5), ("b", 2), ("b", 6), ("c", 3), ("d", 2))) val rdd2 = sc.parallelize(Seq(("a", 10), ("b", 1), ("d", 3))) val rdd3 = sc.parallelize(Seq(("b", 10), ("a", 1))) val result1 = rdd1.cogroup(rdd2).collect() val result2 = rdd1.cogroup(rdd2, rdd3).collect() /_ 执行结果: Array( (d,(CompactBuffer(2),CompactBuffer(3))), (a,(CompactBuffer(1, 2, 5),CompactBuffer(10))), (b,(CompactBuffer(2, 6),CompactBuffer(1))), (c,(CompactBuffer(3),CompactBuffer())) ) _/ println(result1) /_ 执行结果: Array( (d,(CompactBuffer(2),CompactBuffer(3),CompactBuffer())), (a,(CompactBuffer(1, 2, 5),CompactBuffer(10),CompactBuffer(1))), (b,(CompactBuffer(2, 6),CompactBuffer(1),Co... _/ println(result2)作用多个 RDD 协同分组, 将多个 RDD 中 Key 相同的 Value 分组调用cogroup(rdd1, rdd2, rdd3, [partitioner or numPartitions])参数rdd… 最多可以传三个 RDD 进去, 加上调用者, 可以为四个 RDD 协同分组partitioner or numPartitions 可选, 可以通过传递分区函数或者分区数来改变分区注意点对 RDD1, RDD2, RDD3 进行 cogroup, 结果中就一定会有三个 List, 如果没有 Value 则是空 List, 这一点类似于 SQL 的全连接, 返回所有结果, 即使没有关联上CoGroup 是一个需要 Shuffled 的操作
cartesian(other)	(RDD[T], RDD[U]) ⇒ RDD[(T, U)] 生成两个 RDD 的笛卡尔积
sortBy(ascending, numPartitions)	val rdd1 = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1))) val sortByResult = rdd1.sortBy( item => item._2 ).collect() val sortByKeyResult = rdd1.sortByKey().collect() println(sortByResult) println(sortByKeyResult)作用排序相关相关的算子有两个, 一个是 sortBy, 另外一个是 sortByKey调用sortBy(func, ascending, numPartitions)参数func 通过这个函数返回要排序的字段ascending 是否升序numPartitions 分区数注意点普通的 RDD 没有 sortByKey, 只有 Key-Value 的 RDD 才有sortBy 可以指定按照哪个字段来排序, sortByKey 直接按照 Key 来排序
partitionBy(partitioner)	使用用传入的 partitioner 重新分区, 如果和当前分区函数相同, 则忽略操作
coalesce(numPartitions)	减少分区数val rdd = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1))) val oldNum = rdd.partitions.length val coalesceRdd = rdd.coalesce(4, shuffle = true) val coalesceNum = coalesceRdd.partitions.length val repartitionRdd = rdd.repartition(4) val repartitionNum = repartitionRdd.partitions.length print(oldNum, coalesceNum, repartitionNum)作用一般涉及到分区操作的算子常见的有两个, repartitioin 和 coalesce, 两个算子都可以调大或者调小分区数量调用repartitioin(numPartitions)``coalesce(numPartitions, shuffle)参数numPartitions 新的分区数shuffle 是否 shuffle, 如果新的分区数量比原分区数大, 必须 Shuffled, 否则重分区无效注意点repartition 和 coalesce 的不同就在于 coalesce 可以控制是否 Shufflerepartition 是一个 Shuffled 操作
repartition(numPartitions)	重新分区
repartitionAndSortWithinPartitions	重新分区的同时升序排序, 在 partitioner 中排序, 比先重分区再排序要效率高, 建议使用在需要分区后再排序的场景使用

2.2. Action 算子

应用

1
2
3
4
5
6
7
8
9

​````scala
val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
// 结果: Array((手机,10.0), (手机,15.0), (电脑,20.0))
println(rdd.collect())
// 结果: Array((手机,10.0), (手机,15.0))
println(rdd.take(2))
// 结果: (手机,10.0)
println(rdd.first())
​````

总结

RDD 的算子大部分都会生成一些专用的 RDD

1
2

map`, `flatMap`, `filter` 等算子会生成 `MapPartitionsRDD
coalesce`, `repartition` 等算子会生成 `CoalescedRDD

常见的 RDD 有两种类型

转换型的 RDD, Transformation

动作型的 RDD, Action

常见的 Transformation 类型的 RDD

map

flatMap

filter

groupBy

reduceByKey

常见的 Action 类型的 RDD

collect

countByKey

reduce

2.3. RDD 对不同类型数据的支持

目标

1. 理解 RDD 对 Key-Value 类型的数据是有专门支持的
2. 理解 RDD 对数字类型也有专门的支持

一般情况下 RDD 要处理的数据有三类

• 字符串
• 键值对
• 数字型

RDD 的算子设计对这三类不同的数据分别都有支持

对于以字符串为代表的基本数据类型是比较基础的一些的操作, 诸如 map, flatMap, filter 等基础的算子

对于键值对类型的数据, 有额外的支持, 诸如 reduceByKey, groupByKey 等 byKey 的算子

同样对于数字型的数据也有额外的支持, 诸如 max, min 等

RDD 对键值对数据的额外支持

键值型数据本质上就是一个二元元组, 键值对类型的 RDD 表示为 RDD[(K, V)]

RDD 对键值对的额外支持是通过隐式支持来完成的, 一个 RDD[(K, V)], 可以被隐式转换为一个 PairRDDFunctions 对象, 从而调用其中的方法.

既然对键值对的支持是通过 PairRDDFunctions 提供的, 那么从 PairRDDFunctions 中就可以看到这些支持有什么

RDD 对数字型数据的额外支持

对于数字型数据的额外支持基本上都是 Action 操作, 而不是转换操作

1
2
3

`val rdd = sc.parallelize(Seq(1, 2, 3))
// 结果: 3
println(rdd.max())`

2.4. 阶段练习和总结

导读

1. 通过本节, 希望大家能够理解 RDD 的一般使用步骤

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

`// 1. 创建 SparkContext
val conf = new SparkConf().setMaster("local[6]").setAppName("stage_practice1")
val sc = new SparkContext(conf)

// 2. 创建 RDD
val rdd1 = sc.textFile("dataset/BeijingPM20100101_20151231_noheader.csv")
// 3. 处理 RDD
val rdd2 = rdd1.map { item =>
val fields = item.split(",")
((fields(1), fields(2)), fields(6))
}
val rdd3 = rdd2.filter { item => !item._2.equalsIgnoreCase("NA") }
val rdd4 = rdd3.map { item => (item._1, item._2.toInt) }
val rdd5 = rdd4.reduceByKey { (curr, agg) => curr + agg }
val rdd6 = rdd5.sortByKey(ascending = false)
// 4. 行动, 得到结果
println(rdd6.first())`

通过上述代码可以看到, 其实 RDD 的整体使用步骤如下

3. RDD 的 Shuffle 和分区

目标

RDD 的分区操作

Shuffle 的原理

分区的作用

RDD 使用分区来分布式并行处理数据, 并且要做到尽量少的在不同的 Executor 之间使用网络交换数据, 所以当使用 RDD 读取数据的时候, 会尽量的在物理上靠近数据源, 比如说在读取 Cassandra 或者 HDFS 中数据的时候, 会尽量的保持 RDD 的分区和数据源的分区数, 分区模式等一一对应

分区和 Shuffle 的关系

分区的主要作用是用来实现并行计算, 本质上和 Shuffle 没什么关系, 但是往往在进行数据处理的时候, 例如 reduceByKey, groupByKey 等聚合操作, 需要把 Key 相同的 Value 拉取到一起进行计算, 这个时候因为这些 Key 相同的 Value 可能会坐落于不同的分区, 于是理解分区才能理解 Shuffle 的根本原理

Spark 中的 Shuffle 操作的特点

• 只有 Key-Value 型的 RDD 才会有 Shuffle 操作, 例如 RDD[(K, V)], 但是有一个特例, 就是 repartition 算子可以对任何数据类型 Shuffle
• 早期版本 Spark 的 Shuffle 算法是 Hash base shuffle, 后来改为 Sort base shuffle, 更适合大吞吐量的场景

3.1. RDD 的分区操作

查看分区数

1
2

`scala> sc.parallelize(1 to 100).count
res0: Long = 100`

之所以会有 8 个 Tasks, 是因为在启动的时候指定的命令是 spark-shell --master local[8], 这样会生成 1 个 Executors, 这个 Executors 有 8 个 Cores, 所以默认会有 8 个 Tasks, 每个 Cores 对应一个分区, 每个分区对应一个 Tasks, 可以通过 rdd.partitions.size 来查看分区数量

同时也可以通过 spark-shell 的 WebUI 来查看 Executors 的情况

默认的分区数量是和 Cores 的数量有关的, 也可以通过如下三种方式修改或者重新指定分区数量

创建 RDD 时指定分区数

1
2
3
4
5
6
7
8
9

`scala> val rdd1 = sc.parallelize(1 to 100, 6)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:24

scala> rdd1.partitions.size
res1: Int = 6
scala> val rdd2 = sc.textFile("hdfs:///dataset/wordcount.txt", 6)
rdd2: org.apache.spark.rdd.RDD[String] = hdfs:///dataset/wordcount.txt MapPartitionsRDD[3] at textFile at <console>:24
scala> rdd2.partitions.size
res2: Int = 7`

rdd1 是通过本地集合创建的, 创建的时候通过第二个参数指定了分区数量. rdd2 是通过读取 HDFS 中文件创建的, 同样通过第二个参数指定了分区数, 因为是从 HDFS 中读取文件, 所以最终的分区数是由 Hadoop 的 InputFormat 来指定的, 所以比指定的分区数大了一个.

通过 coalesce 算子指定

1

`coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]`

numPartitions

新生成的 RDD 的分区数

shuffle

是否 Shuffle

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

`scala> val source = sc.parallelize(1 to 100, 6)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> source.partitions.size
res0: Int = 6
scala> val noShuffleRdd = source.coalesce(numPartitions=8, shuffle=false)
noShuffleRdd: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[1] at coalesce at <console>:26
scala> noShuffleRdd.toDebugString (1)
res1: String =
(6) CoalescedRDD[1] at coalesce at <console>:26 []
| ParallelCollectionRDD[0] at parallelize at <console>:24 []
scala> val noShuffleRdd = source.coalesce(numPartitions=8, shuffle=false)
noShuffleRdd: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[1] at coalesce at <console>:26
scala> shuffleRdd.toDebugString (2)
res3: String =
(8) MapPartitionsRDD[5] at coalesce at <console>:26 []
| CoalescedRDD[4] at coalesce at <console>:26 []
| ShuffledRDD[3] at coalesce at <console>:26 []
+-(6) MapPartitionsRDD[2] at coalesce at <console>:26 []
| ParallelCollectionRDD[0] at parallelize at <console>:24 []
scala> noShuffleRdd.partitions.size (3)
res4: Int = 6
scala> shuffleRdd.partitions.size
res5: Int = 8`

通过 repartition 算子指定

1
2

`repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]`
repartition` 算子本质上就是 `coalesce(numPartitions, shuffle = true)

1
2
3
4
5
6
7
8
9

`scala> val source = sc.parallelize(1 to 100, 6)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[7] at parallelize at <console>:24

scala> source.partitions.size
res7: Int = 6
scala> source.repartition(100).partitions.size (1)
res8: Int = 100
scala> source.repartition(1).partitions.size (2)
res9: Int = 1`

repartition 算子无论是增加还是减少分区都是有效的, 因为本质上 repartition 会通过 shuffle 操作把数据分发给新的 RDD 的不同的分区, 只有 shuffle 操作才可能做到增大分区数, 默认情况下, 分区函数是 RoundRobin, 如果希望改变分区函数, 也就是数据分布的方式, 可以通过自定义分区函数来实现

3.2. RDD 的 Shuffle 是什么

1
2
3
4

`val sourceRdd = sc.textFile("hdfs://node01:9020/dataset/wordcount.txt")
val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_, 1))
val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
val result = aggCountRdd.collect`

reduceByKey 这个算子本质上就是先按照 Key 分组, 后对每一组数据进行 reduce, 所面临的挑战就是 Key 相同的所有数据可能分布在不同的 Partition 分区中, 甚至可能在不同的节点中, 但是它们必须被共同计算.

为了让来自相同 Key 的所有数据都在 reduceByKey 的同一个 reduce 中处理, 需要执行一个 all-to-all 的操作, 需要在不同的节点(不同的分区)之间拷贝数据, 必须跨分区聚集相同 Key 的所有数据, 这个过程叫做 Shuffle.

3.3. RDD 的 Shuffle 原理

Spark 的 Shuffle 发展大致有两个阶段: Hash base shuffle 和 Sort base shuffle

Hash base shuffle

大致的原理是分桶, 假设 Reducer 的个数为 R, 那么每个 Mapper 有 R 个桶, 按照 Key 的 Hash 将数据映射到不同的桶中, Reduce 找到每一个 Mapper 中对应自己的桶拉取数据.

假设 Mapper 的个数为 M, 整个集群的文件数量是 M * R, 如果有 1,000 个 Mapper 和 Reducer, 则会生成 1,000,000 个文件, 这个量非常大了.

过多的文件会导致文件系统打开过多的文件描述符, 占用系统资源. 所以这种方式并不适合大规模数据的处理, 只适合中等规模和小规模的数据处理, 在 Spark 1.2 版本中废弃了这种方式.

Sort base shuffle

对于 Sort base shuffle 来说, 每个 Map 侧的分区只有一个输出文件, Reduce 侧的 Task 来拉取, 大致流程如下

1. Map 侧将数据全部放入一个叫做 AppendOnlyMap 的组件中, 同时可以在这个特殊的数据结构中做聚合操作

2. 然后通过一个类似于 MergeSort 的排序算法 TimSort 对 AppendOnlyMap 底层的 Array 排序

   1	* 先按照 Partition ID 排序, 后按照 Key 的 HashCode 排序

3. 最终每个 Map Task 生成一个 输出文件, Reduce Task 来拉取自己对应的数据

从上面可以得到结论, Sort base shuffle 确实可以大幅度减少所产生的中间文件, 从而能够更好的应对大吞吐量的场景, 在 Spark 1.2 以后, 已经默认采用这种方式.

但是需要大家知道的是, Spark 的 Shuffle 算法并不只是这一种, 即使是在最新版本, 也有三种 Shuffle 算法, 这三种算法对每个 Map 都只产生一个临时文件, 但是产生文件的方式不同, 一种是类似 Hash 的方式, 一种是刚才所说的 Sort, 一种是对 Sort 的一种优化(使用 Unsafe API 直接申请堆外内存)

4. 缓存

概要

缓存的意义

缓存相关的 API

缓存级别以及最佳实践

4.1. 缓存的意义

使用缓存的原因 - 多次使用 RDD

需求: 在日志文件中找到访问次数最少的 IP 和访问次数最多的 IP

1
2
3
4
5
6
7
8
9
10
11

`val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg) (1)
val resultLess = interimRDD.sortBy(item => item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item => item._2, ascending = false).first()
println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")
sc.stop()`

在上述代码中, 多次使用到了 interimRDD, 导致文件读取两次, 计算两次, 有没有什么办法增进上述代码的性能?

使用缓存的原因 - 容错

当在计算 RDD3 的时候如果出错了, 会怎么进行容错?

会再次计算 RDD1 和 RDD2 的整个链条, 假设 RDD1 和 RDD2 是通过比较昂贵的操作得来的, 有没有什么办法减少这种开销?

上述两个问题的解决方案其实都是 缓存, 除此之外, 使用缓存的理由还有很多, 但是总结一句, 就是缓存能够帮助开发者在进行一些昂贵操作后, 将其结果保存下来, 以便下次使用无需再次执行, 缓存能够显著的提升性能.

所以, 缓存适合在一个 RDD 需要重复多次利用, 并且还不是特别大的情况下使用, 例如迭代计算等场景.

4.2. 缓存相关的 API

可以使用 cache 方法进行缓存

1
2
3
4
5
6
7
8
9
10
11
12

`val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg)
.cache() (1)
val resultLess = interimRDD.sortBy(item => item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item => item._2, ascending = false).first()
println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")
sc.stop()`

方法签名如下

1

`cache(): this.type = persist()`

cache 方法其实是 persist 方法的一个别名

也可以使用 persist 方法进行缓存

1
2
3
4
5
6
7
8
9
10
11
12

val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg)
.persist(StorageLevel.MEMORY_ONLY) (1)
val resultLess = interimRDD.sortBy(item => item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item => item._2, ascending = false).first()
println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")
sc.stop()

方法签名如下

1
2

`persist(): this.type
persist(newLevel: StorageLevel): this.type`

persist 方法其实有两种形式, persist() 是 persist(newLevel: StorageLevel) 的一个别名, persist(newLevel: StorageLevel) 能够指定缓存的级别

缓存其实是一种空间换时间的做法, 会占用额外的存储资源, 如何清理?

1
2
3
4
5
6
7
8
9
10
11
12
13

`val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg)
.persist()
interimRDD.unpersist() (1)
val resultLess = interimRDD.sortBy(item => item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item => item._2, ascending = false).first()
println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")
sc.stop()`

根据缓存级别的不同, 缓存存储的位置也不同, 但是使用 unpersist 可以指定删除 RDD 对应的缓存信息, 并指定缓存级别为 NONE

4.3. 缓存级别

其实如何缓存是一个技术活, 有很多细节需要思考, 如下

• 是否使用磁盘缓存?
• 是否使用内存缓存?
• 是否使用堆外内存?
• 缓存前是否先序列化?
• 是否需要有副本?

如果要回答这些信息的话, 可以先查看一下 RDD 的缓存级别对象

1
2
3
4
5
6
7
8
9
10

`val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg)
.persist()
println(interimRDD.getStorageLevel)
sc.stop()`

打印出来的对象是 StorageLevel, 其中有如下几个构造参数

根据这几个参数的不同, StorageLevel 有如下几个枚举对象

如何选择分区级别

Spark 的存储级别的选择，核心问题是在 memory 内存使用率和 CPU 效率之间进行权衡。建议按下面的过程进行存储级别的选择:

如果您的 RDD 适合于默认存储级别（MEMORY_ONLY），leave them that way。这是 CPU 效率最高的选项，允许 RDD 上的操作尽可能快地运行.

如果不是，试着使用 MEMORY_ONLY_SER 和 selecting a fast serialization library 以使对象更加节省空间，但仍然能够快速访问。(Java 和 Scala)

不要溢出到磁盘，除非计算您的数据集的函数是昂贵的，或者它们过滤大量的数据。否则，重新计算分区可能与从磁盘读取分区一样快.

如果需要快速故障恢复，请使用复制的存储级别（例如，如果使用 Spark 来服务 来自网络应用程序的请求）。All 存储级别通过重新计算丢失的数据来提供完整的容错能力，但复制的数据可让您继续在 RDD 上运行任务，而无需等待重新计算一个丢失的分区.

5. Checkpoint

目标

1. Checkpoint 的作用
2. Checkpoint 的使用

5.1. Checkpoint 的作用

Checkpoint 的主要作用是斩断 RDD 的依赖链, 并且将数据存储在可靠的存储引擎中, 例如支持分布式存储和副本机制的 HDFS.

Checkpoint 的方式

• 可靠的 将数据存储在可靠的存储引擎中, 例如 HDFS
• 本地的 将数据存储在本地

什么是斩断依赖链

斩断依赖链是一个非常重要的操作, 接下来以 HDFS 的 NameNode 的原理来举例说明

HDFS 的 NameNode 中主要职责就是维护两个文件, 一个叫做 edits, 另外一个叫做 fsimage. edits 中主要存放 EditLog, FsImage 保存了当前系统中所有目录和文件的信息. 这个 FsImage 其实就是一个 Checkpoint.

HDFS 的 NameNode 维护这两个文件的主要过程是, 首先, 会由 fsimage 文件记录当前系统某个时间点的完整数据, 自此之后的数据并不是时刻写入 fsimage, 而是将操作记录存储在 edits 文件中. 其次, 在一定的触发条件下, edits 会将自身合并进入 fsimage. 最后生成新的 fsimage 文件, edits 重置, 从新记录这次 fsimage 以后的操作日志.

如果不合并 edits 进入 fsimage 会怎样? 会导致 edits 中记录的日志过长, 容易出错.

所以当 Spark 的一个 Job 执行流程过长的时候, 也需要这样的一个斩断依赖链的过程, 使得接下来的计算轻装上阵.

Checkpoint 和 Cache 的区别

Cache 可以把 RDD 计算出来然后放在内存中, 但是 RDD 的依赖链(相当于 NameNode 中的 Edits 日志)是不能丢掉的, 因为这种缓存是不可靠的, 如果出现了一些错误(例如 Executor 宕机), 这个 RDD 的容错就只能通过回溯依赖链, 重放计算出来.

但是 Checkpoint 把结果保存在 HDFS 这类存储中, 就是可靠的了, 所以可以斩断依赖, 如果出错了, 则通过复制 HDFS 中的文件来实现容错.

所以他们的区别主要在以下两点

Checkpoint 可以保存数据到 HDFS 这类可靠的存储上, Persist 和 Cache 只能保存在本地的磁盘和内存中

Checkpoint 可以斩断 RDD 的依赖链, 而 Persist 和 Cache 不行

因为 CheckpointRDD 没有向上的依赖链, 所以程序结束后依然存在, 不会被删除. 而 Cache 和 Persist 会在程序结束后立刻被清除.

5.2. 使用 Checkpoint

1
2
3
4
5
6
7
8
9
10
11

`val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)
sc.setCheckpointDir("checkpoint") **(1)**

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
.map(item => (item.split(" ")(0), 1))
.filter(item => StringUtils.isNotBlank(item._1))
.reduceByKey((curr, agg) => curr + agg)
interimRDD.checkpoint() (2)
interimRDD.collect().foreach(println(_))
sc.stop()`

6. Spark 底层逻辑

导读

1. 从部署图了解 Spark 部署了什么, 有什么组件运行在集群中
2. 通过对 WordCount 案例的解剖, 来理解执行逻辑计划的生成
3. 通过对逻辑执行计划的细化, 理解如何生成物理计划

部署情况

在 Spark 部分的底层执行逻辑开始之前, 还是要先认识一下 Spark 的部署情况, 根据部署情况, 从而理解如何调度.

针对于上图, 首先可以看到整体上在集群中运行的角色有如下几个:

• Master Daemon

  负责管理 Master 节点, 协调资源的获取, 以及连接 Worker 节点来运行 Executor, 是 Spark 集群中的协调节点

• Worker Daemon

  Workers 也称之为叫 Slaves, 是 Spark 集群中的计算节点, 用于和 Master 交互并管理 Executor.

  当一个 Spark Job 提交后, 会创建 SparkContext, 后 Worker 会启动对应的 Executor.

• Executor Backend

  上面有提到 Worker 用于控制 Executor 的启停, 其实 Worker 是通过 Executor Backend 来进行控制的, Executor Backend 是一个进程(是一个 JVM 实例), 持有一个 Executor 对象

另外在启动程序的时候, 有三种程序需要运行在集群上:

1

Driver

Driver 是一个 JVM 实例, 是一个进程, 是 Spark Application 运行时候的领导者, 其中运行了 SparkContext.

Driver 控制 Job 和 Task, 并且提供 WebUI.

1

Executor

Executor 对象中通过线程池来运行 Task, 一个 Executor 中只会运行一个 Spark Application 的 Task, 不同的 Spark Application 的 Task 会由不同的 Executor 来运行

案例

因为要理解执行计划, 重点不在案例, 所以本节以一个非常简单的案例作为入门, 就是我们第一个案例 WordCount

1
2
3
4
5
6
7
8
9

`val sc = ...

val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD = textRDD.flatMap(.split(" "))
val tupleRDD = splitRDD.map((, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)
val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")
println(strRDD.toDebugString)
strRDD.collect.foreach(item => println(item))`

整个案例的运行过程大致如下:

1. 通过代码的运行, 生成对应的 RDD 逻辑执行图
2. 通过 Action 操作, 根据逻辑执行图生成对应的物理执行图, 也就是 Stage 和 Task
3. 将物理执行图运行在集群中

逻辑执行图

对于上面代码中的 reduceRDD 如果使用 toDebugString 打印调试信息的话, 会显式如下内容

1
2
3
4
5

`(6) MapPartitionsRDD[4] at map at WordCount.scala:20 []
 |  ShuffledRDD[3] at reduceByKey at WordCount.scala:19 []
 +-(6) MapPartitionsRDD[2] at map at WordCount.scala:18 []
    |  MapPartitionsRDD[1] at flatMap at WordCount.scala:17 []
    |  ParallelCollectionRDD[0] at parallelize at WordCount.scala:16 []`

根据这段内容, 大致能得到这样的一张逻辑执行图

其实 RDD 并没有什么严格的逻辑执行图和物理执行图的概念, 这里也只是借用这个概念, 从而让整个 RDD 的原理可以解释, 好理解.

对于 RDD 的逻辑执行图, 起始于第一个入口 RDD 的创建, 结束于 Action 算子执行之前, 主要的过程就是生成一组互相有依赖关系的 RDD, 其并不会真的执行, 只是表示 RDD 之间的关系, 数据的流转过程.

物理执行图

当触发 Action 执行的时候, 这一组互相依赖的 RDD 要被处理, 所以要转化为可运行的物理执行图, 调度到集群中执行.

因为大部分 RDD 是不真正存放数据的, 只是数据从中流转, 所以, 不能直接在集群中运行 RDD, 要有一种 Pipeline 的思想, 需要将这组 RDD 转为 Stage 和 Task, 从而运行 Task, 优化整体执行速度.

以上的逻辑执行图会生成如下的物理执行图, 这一切发生在 Action 操作被执行时.

从上图可以总结如下几个点

• 在第一个 Stage 中, 每一个这样的执行流程是一个 Task, 也就是在同一个 Stage 中的所有 RDD 的对应分区, 在同一个 Task 中执行
• Stage 的划分是由 Shuffle 操作来确定的, 有 Shuffle 的地方, Stage 断开

6.1. 逻辑执行图生成

导读

1. 如何生成 RDD
2. 如何控制 RDD 之间的关系

6.1.1. RDD 的生成

重点内容

本章要回答如下三个问题

• 如何生成 RDD
• 生成什么 RDD
• 如何计算 RDD 中的数据

1
2
3
4
5
6
7
8
9

`val sc = ...

val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD = textRDD.flatMap(.split(" "))
val tupleRDD = splitRDD.map((, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)
val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")
println(strRDD.toDebugString)
strRDD.collect.foreach(item => println(item))`

明确逻辑计划的边界

在 Action 调用之前, 会生成一系列的 RDD, 这些 RDD 之间的关系, 其实就是整个逻辑计划

例如上述代码, 如果生成逻辑计划的, 会生成如下一些 RDD, 这些 RDD 是相互关联的, 这些 RDD 之间, 其实本质上生成的就是一个 计算链

接下来, 采用迭代渐进式的方式, 一步一步的查看一下整体上的生成过程

textFile 算子的背后

研究 RDD 的功能或者表现的时候, 其实本质上研究的就是 RDD 中的五大属性, 因为 RDD 透过五大属性来提供功能和表现, 所以如果要研究 textFile 这个算子, 应该从五大属性着手, 那么第一步就要看看生成的 RDD 是什么类型的 RDD

1. textFile
   1
   2
   3
   4
   5
   6
   7
   
    
   
   生成的是
   
    
   
   
   HadoopRDD
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   
      ![20190519202310](https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190519202310.png)
   
      ![20190519202411](https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190519202411.png)
   
      |      | 除了上面这一个步骤以外, 后续步骤将不再直接基于代码进行讲解, 因为从代码的角度着手容易迷失逻辑, 这个章节的初心有两个, 一个是希望大家了解 Spark 的内部逻辑和原理, 另外一个是希望大家能够通过本章学习具有代码分析的能力 |
      | ---- | ------------------------------------------------------------ |
      |      |                                                              |
   
   2. ```
      HadoopRDD
   
   
   
   

的

1

Partitions

对应了

1

HDFS

的

1

Blocks

其实本质上每个 HadoopRDD 的 Partition 都是对应了一个 Hadoop 的 Block, 通过 InputFormat 来确定 Hadoop 中的 Block 的位置和边界, 从而可以供一些算子使用

3. HadoopRDD
   1
   2
   3
   4
   5
   6
   7
   
    
   
   的
   
    
   
   
   compute
   1
   2
   3
   4
   5
   6
   7
   
    
   
   函数就是在读取
   
    
   
   
   HDFS
   1
   2
   3
   4
   5
   6
   7
   
    
   
   中的
   
    
   
   
   Block
   1
   2
   3
   4
   5
   
      本质上, `compute` 还是依然使用 `InputFormat` 来读取 `HDFS` 中对应分区的 `Block`
   
   4. ```
      textFile
   
   
   
   

这个算子生成的其实是一个

1

MapPartitionsRDD

textFile 这个算子的作用是读取 HDFS 上的文件, 但是 HadoopRDD 中存放是一个元组, 其 Key 是行号, 其 Value 是 Hadoop 中定义的 Text 对象, 这一点和 MapReduce 程序中的行为是一致的

但是并不适合 Spark 的场景, 所以最终会通过一个 map 算子, 将 (LineNum, Text) 转为 String 形式的一行一行的数据, 所以最终 textFile 这个算子生成的 RDD 并不是 HadoopRDD, 而是一个 MapPartitionsRDD

map 算子的背后

• map 算子生成了 MapPartitionsRDD

  由源码可知, 当 val rdd2 = rdd1.map() 的时候, 其实生成的新 RDD 是 rdd2, rdd2 的类型是 MapPartitionsRDD, 每个 RDD 中的五大属性都会有一些不同, 由 map 算子生成的 RDD 中的计算函数, 本质上就是遍历对应分区的数据, 将每一个数据转成另外的形式

• MapPartitionsRDD 的计算函数是 collection.map( function )

  真正运行的集群中的处理单元是 Task, 每个 Task 对应一个 RDD 的分区, 所以 collection 对应一个 RDD 分区的所有数据, 而这个计算的含义就是将一个 RDD 的分区上所有数据当作一个集合, 通过这个 Scala 集合的 map 算子, 来执行一个转换操作, 其转换操作的函数就是传入 map 算子的 function

• 传入 map 算子的函数会被清理

  

  这个清理主要是处理闭包中的依赖, 使得这个闭包可以被序列化发往不同的集群节点运行

flatMap 算子的背后

1

flatMap` 和 `map` 算子其实本质上是一样的, 其步骤和生成的 `RDD` 都是一样, 只是对于传入函数的处理不同, `map` 是 `collect.map( function )` 而 `flatMap` 是 `collect.flatMap( function )

从侧面印证了, 其实 Spark 中的 flatMap 和 Scala 基础中的 flatMap 其实是一样的

1

textRDD` → `splitRDD` → `tupleRDD

由 textRDD 到 splitRDD 再到 tupleRDD 的过程, 其实就是调用 map 和 flatMap 算子生成新的 RDD 的过程, 所以如下图所示, 就是这个阶段所生成的逻辑计划

总结

如何生成 RDD ?

生成 RDD 的常见方式有三种

从本地集合创建

从外部数据集创建

从其它 RDD 衍生

通过外部数据集创建 RDD, 是通过 Hadoop 或者其它外部数据源的 SDK 来进行数据读取, 同时如果外部数据源是有分片的话, RDD 会将分区与其分片进行对照

通过其它 RDD 衍生的话, 其实本质上就是通过不同的算子生成不同的 RDD 的子类对象, 从而控制 compute 函数的行为来实现算子功能

生成哪些 RDD ?

不同的算子生成不同的 RDD, 生成 RDD 的类型取决于算子, 例如 map 和 flatMap 都会生成 RDD 的子类 MapPartitions 的对象

如何计算 RDD 中的数据 ?

虽然前面我们提到过 RDD 是偏向计算的, 但是其实 RDD 还只是表示数据, 纵观 RDD 的五大属性中有三个是必须的, 分别如下

Partitions List 分区列表

Compute function 计算函数

Dependencies 依赖

虽然计算函数是和计算有关的, 但是只有调用了这个函数才会进行计算, RDD 显然不会自己调用自己的 Compute 函数, 一定是由外部调用的, 所以 RDD 更多的意义是用于表示数据集以及其来源, 和针对于数据的计算

所以如何计算 RDD 中的数据呢? 一定是通过其它的组件来计算的, 而计算的规则, 由 RDD 中的 Compute 函数来指定, 不同类型的 RDD 子类有不同的 Compute 函数

6.1.2. RDD 之间的依赖关系

导读

1. 讨论什么是 RDD 之间的依赖关系
2. 继而讨论 RDD 分区之间的关系
3. 最后确定 RDD 之间的依赖关系分类
4. 完善案例的逻辑关系图

什么是 RDD 之间的依赖关系?

• 什么是关系(依赖关系) ?

  从算子视角上来看, splitRDD 通过 map 算子得到了 tupleRDD, 所以 splitRDD 和 tupleRDD 之间的关系是 map

  但是仅仅这样说, 会不够全面, 从细节上来看, RDD 只是数据和关于数据的计算, 而具体执行这种计算得出结果的是一个神秘的其它组件, 所以, 这两个 RDD 的关系可以表示为 splitRDD 的数据通过 map 操作, 被传入 tupleRDD, 这是它们之间更细化的关系

  但是 RDD 这个概念本身并不是数据容器, 数据真正应该存放的地方是 RDD 的分区, 所以如果把视角放在数据这一层面上的话, 直接讲这两个 RDD 之间有关系是不科学的, 应该从这两个 RDD 的分区之间的关系来讨论它们之间的关系

• 那这些分区之间是什么关系?

  如果仅仅说 splitRDD 和 tupleRDD 之间的话, 那它们的分区之间就是一对一的关系

  但是 tupleRDD 到 reduceRDD 呢? tupleRDD 通过算子 reduceByKey 生成 reduceRDD, 而这个算子是一个 Shuffle 操作, Shuffle 操作的两个 RDD 的分区之间并不是一对一, reduceByKey 的一个分区对应 tupleRDD 的多个分区

1

reduceByKey` 算子会生成 `ShuffledRDD

reduceByKey 是由算子 combineByKey 来实现的, combineByKey 内部会创建 ShuffledRDD 返回, 具体的代码请大家通过 IDEA 来进行查看, 此处不再截图, 而整个 reduceByKey 操作大致如下过程

去掉两个 reducer 端的分区, 只留下一个的话, 如下

所以, 对于 reduceByKey 这个 Shuffle 操作来说, reducer 端的一个分区, 会从多个 mapper 端的分区拿取数据, 是一个多对一的关系

至此为止, 出现了两种分区见的关系了, 一种是一对一, 一种是多对一

整体上的流程图

6.1.3. RDD 之间的依赖关系详解

导读

上个小节通过例子演示了 RDD 的分区间的关系有两种形式

一对一, 一般是直接转换

多对一, 一般是 Shuffle

本小节会说明如下问题:

1. 如果分区间得关系是一对一或者多对一, 那么这种情况下的 RDD 之间的关系的正式命名是什么呢?
2. RDD 之间的依赖关系, 具体有几种情况呢?

窄依赖

假如 rddB = rddA.transform(…), 如果 rddB 中一个分区依赖 rddA 也就是其父 RDD 的少量分区, 这种 RDD 之间的依赖关系称之为窄依赖

换句话说, 子 RDD 的每个分区依赖父 RDD 的少量个数的分区, 这种依赖关系称之为窄依赖

举个栗子

1
2
3
4
5
6
7
8
9

`val sc = ...

val rddA = sc.parallelize(Seq(1, 2, 3))
val rddB = sc.parallelize(Seq("a", "b"))
/**

运行结果: (1,a), (1,b), (2,a), (2,b), (3,a), (3,b)
*/
rddA.cartesian(rddB).collect().foreach(println(_))`

• 上述代码的 cartesian 是求得两个集合的笛卡尔积
• 上述代码的运行结果是 rddA 中每个元素和 rddB 中的所有元素结合, 最终的结果数量是两个 RDD 数量之和
• rddC 有两个父 RDD, 分别为 rddA 和 rddB

对于 cartesian 来说, 依赖关系如下

上述图形中清晰展示如下现象

1. rddC 中的分区数量是两个父 RDD 的分区数量之乘积
2. rddA 中每个分区对应 rddC 中的两个分区 (因为 rddB 中有两个分区), rddB 中的每个分区对应 rddC 中的三个分区 (因为 rddA 有三个分区)

它们之间是窄依赖, 事实上在 cartesian 中也是 NarrowDependency 这个所有窄依赖的父类的唯一一次直接使用, 为什么呢?

因为所有的分区之间是拷贝关系, 并不是 Shuffle 关系

• rddC 中的每个分区并不是依赖多个父 RDD 中的多个分区
• rddC 中每个分区的数量来自一个父 RDD 分区中的所有数据, 是一个 FullDependence, 所以数据可以直接从父 RDD 流动到子 RDD
• 不存在一个父 RDD 中一部分数据分发过去, 另一部分分发给其它的 RDD

宽依赖

在 ShuffleDependency 的类声明上如下写到

1

`Represents a dependency on the output of a shuffle stage.`

上面非常清楚的说道, 宽依赖就是 Shuffle 中的依赖关系, 换句话说, 只有 Shuffle 产生的地方才是宽依赖

那么宽窄依赖的判断依据就非常简单明确了, 是否有 Shuffle ?

举个 reduceByKey 的例子, rddB = rddA.reduceByKey( (curr, agg) ⇒ curr + agg ) 会产生如下的依赖关系

rddB 的每个分区都几乎依赖 rddA 的所有分区

对于 rddA 中的一个分区来说, 其将一部分分发给 rddB 的 p1, 另外一部分分发给 rddB 的 p2, 这不是数据流动, 而是分发

如何分辨宽窄依赖 ?

其实分辨宽窄依赖的本身就是在分辨父子 RDD 之间是否有 Shuffle, 大致有以下的方法

如果是 Shuffle, 两个 RDD 的分区之间不是单纯的数据流动, 而是分发和复制

一般 Shuffle 的子 RDD 的每个分区会依赖父 RDD 的多个分区

但是这样判断其实不准确, 如果想分辨某个算子是否是窄依赖, 或者是否是宽依赖, 则还是要取决于具体的算子, 例如想看 cartesian 生成的是宽依赖还是窄依赖, 可以通过如下步骤

1. 查看

1

map

算子生成的

1

RDD

2. 进去

1

RDD

查看

1

getDependence

方法

总结

• RDD 的逻辑图本质上是对于计算过程的表达, 例如数据从哪来, 经历了哪些步骤的计算
• 每一个步骤都对应一个 RDD, 因为数据处理的情况不同, RDD 之间的依赖关系又分为窄依赖和宽依赖 *

6.1.4. 常见的窄依赖类型

导读

常见的窄依赖其实也是有分类的, 而且宽窄以来不太容易分辨, 所以通过本章, 帮助同学明确窄依赖的类型

一对一窄依赖

其实 RDD 中默认的是 OneToOneDependency, 后被不同的 RDD 子类指定为其它的依赖类型, 常见的一对一依赖是 map 算子所产生的依赖, 例如 rddB = rddA.map(…)

每个分区之间一一对应, 所以叫做一对一窄依赖

Range 窄依赖

1

Range` 窄依赖其实也是一对一窄依赖, 但是保留了中间的分隔信息, 可以通过某个分区获取其父分区, 目前只有一个算子生成这种窄依赖, 就是 `union` 算子, 例如 `rddC = rddA.union(rddB)

1

rddC` 其实就是 `rddA` 拼接 `rddB` 生成的, 所以 `rddC` 的 `p5` 和 `p6` 就是 `rddB` 的 `p1` 和 `p2

所以需要有方式获取到 rddC 的 p5 其父分区是谁, 于是就需要记录一下边界, 其它部分和一对一窄依赖一样

多对一窄依赖

多对一窄依赖其图形和 Shuffle 依赖非常相似, 所以在遇到的时候, 要注意其 RDD 之间是否有 Shuffle 过程, 比较容易让人困惑, 常见的多对一依赖就是重分区算子 coalesce, 例如 rddB = rddA.coalesce(2, shuffle = false), 但同时也要注意, 如果 shuffle = true 那就是完全不同的情况了

因为没有 Shuffle, 所以这是一个窄依赖

再谈宽窄依赖的区别

宽窄依赖的区别非常重要, 因为涉及了一件非常重要的事情: 如何计算 RDD ?

宽窄以来的核心区别是: 窄依赖的 RDD 可以放在一个 Task 中运行

6.2. 物理执行图生成

1. 物理图的意义
2. 如何划分 Task
3. 如何划分 Stage

物理图的作用是什么?

问题一: 物理图的意义是什么?

物理图解决的其实就是 RDD 流程生成以后, 如何计算和运行的问题, 也就是如何把 RDD 放在集群中执行的问题

问题二: 如果要确定如何运行的问题, 则需要先确定集群中有什么组件

• 首先集群中物理元件就是一台一台的机器

• 其次这些机器上跑的守护进程有两种: Master, Worker

  1	* 每个守护进程其实就代表了一台机器, 代表这台机器的角色, 代表这台机器和外界通信

  • 例如我们常说一台机器是 Master, 其含义是这台机器中运行了一个 Master 守护进程, 如果一台机器运行了 Master 的同时又运行了 Worker, 则说这台机器是 Master 也可以, 说它是 Worker 也行

真正能运行 RDD 的组件是: Executor, 也就是说其实 RDD 最终是运行在 Executor 中的, 也就是说, 无论是 Master 还是 Worker 其实都是用于管理 Executor 和调度程序的

结论是 RDD 一定在 Executor 中计算, 而 Master 和 Worker 负责调度和管理 Executor

问题三: 物理图的生成需要考虑什么问题?

要计算 RDD, 不仅要计算, 还要很快的计算 → 优化性能

要考虑容错, 容错的常见手段是缓存 → RDD 要可以缓存

结论是在生成物理图的时候, 不仅要考虑效率问题, 还要考虑一种更合适的方式, 让 RDD 运行的更好

谁来计算 RDD ?

问题一: RDD 是什么, 用来做什么 ?

回顾一下 RDD 的五个属性

1
2
3
4
5

A list of partitions
A function for computing each split
A list of dependencies on other RDDs
Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

简单的说就是: 分区列表, 计算函数, 依赖关系, 分区函数, 最佳位置

分区列表, 分区函数, 最佳位置, 这三个属性其实说的就是数据集在哪, 在哪更合适, 如何分区

计算函数和依赖关系, 这两个属性其实说的是数据集从哪来

所以结论是 RDD 是一个数据集的表示, 不仅表示了数据集, 还表示了这个数据集从哪来, 如何计算

但是问题是, 谁来计算 ? 如果为一台汽车设计了一个设计图, 那么设计图自己生产汽车吗 ?

问题二: 谁来计算 ?

前面我们明确了两件事, RDD 在哪被计算? 在 Executor 中. RDD 是什么? 是一个数据集以及其如何计算的图纸.

直接使用 Executor 也是不合适的, 因为一个计算的执行总是需要一个容器, 例如 JVM 是一个进程, 只有进程中才能有线程, 所以这个计算 RDD 的线程应该运行在一个进程中, 这个进程就是 Exeutor, Executor 有如下两个职责

和 Driver 保持交互从而认领属于自己的任务

接受任务后, 运行任务

所以, 应该由一个线程来执行 RDD 的计算任务, 而 Executor 作为执行这个任务的容器, 也就是一个进程, 用于创建和执行线程, 这个执行具体计算任务的线程叫做 Task

问题三: Task 该如何设计 ?

第一个想法是每个 RDD 都由一个 Task 来计算 第二个想法是一整个逻辑执行图中所有的 RDD 都由一组 Task 来执行 第三个想法是分阶段执行

第一个想法: 为每个 RDD 的分区设置一组 Task

大概就是每个 RDD 都有三个 Task, 每个 Task 对应一个 RDD 的分区, 执行一个分区的数据的计算

但是这么做有一个非常难以解决的问题, 就是数据存储的问题, 例如 Task 1, 4, 7, 10, 13, 16 在同一个流程上, 但是这些 Task 之间需要交换数据, 因为这些 Task 可能被调度到不同的机器上上, 所以 Task1 执行完了数据以后需要暂存, 后交给 Task4 来获取

这只是一个简单的逻辑图, 如果是一个复杂的逻辑图, 会有什么表现? 要存储多少数据? 无论是放在磁盘还是放在内存中, 是不是都是一种极大的负担?

第二个想法: 让数据流动

很自然的, 第一个想法的问题是数据需要存储和交换, 那不存储不就好了吗? 对, 可以让数据流动起来

第一个要解决的问题就是, 要为数据创建管道(Pipeline), 有了管道, 就可以流动

简单来说, 就是为所有的 RDD 有关联的分区使用同一个 Task, 但是就没问题了吗? 请关注红框部分

这两个 RDD 之间是 Shuffle 关系, 也就是说, 右边的 RDD 的一个分区可能依赖左边 RDD 的所有分区, 这样的话, 数据在这个地方流不动了, 怎么办?

第三个想法: 划分阶段

既然在 Shuffle 处数据流不动了, 那就可以在这个地方中断一下, 后面 Stage 部分详解

如何划分阶段 ?

为了减少执行任务, 减少数据暂存和交换的机会, 所以需要创建管道, 让数据沿着管道流动, 其实也就是原先每个 RDD 都有一组 Task, 现在改为所有的 RDD 共用一组 Task, 但是也有问题, 问题如下

就是说, 在 Shuffle 处, 必须断开管道, 进行数据交换, 交换过后, 继续流动, 所以整个流程可以变为如下样子

把 Task 断开成两个部分, Task4 可以从 Task 1, 2, 3 中获取数据, 后 Task4 又作为管道, 继续让数据在其中流动

但是还有一个问题, 说断开就直接断开吗? 不用打个招呼的呀? 这个断开即没有道理, 也没有规则, 所以可以为这个断开增加一个概念叫做阶段, 按照阶段断开, 阶段的英文叫做 Stage, 如下

所以划分阶段的本身就是设置断开点的规则, 那么该如何划分阶段呢?

1. 第一步, 从最后一个 RDD, 也就是逻辑图中最右边的 RDD 开始, 向前滑动 Stage 的范围, 为 Stage0
2. 第二步, 遇到 ShuffleDependency 断开 Stage, 从下一个 RDD 开始创建新的 Stage, 为 Stage1
3. 第三步, 新的 Stage 按照同样的规则继续滑动, 直到包裹所有的 RDD

总结来看, 就是针对于宽窄依赖来判断, 一个 Stage 中只有窄依赖, 因为只有窄依赖才能形成数据的 Pipeline.

如果要进行 Shuffle 的话, 数据是流不过去的, 必须要拷贝和拉取. 所以遇到 RDD 宽依赖的两个 RDD 时, 要切断这两个 RDD 的 Stage.

数据怎么流动 ?

1
2
3
4
5
6
7
8

`val sc = ...

val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD = textRDD.flatMap(.split(" "))
val tupleRDD = splitRDD.map((, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)
val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")
strRDD.collect.foreach(item => println(item))`

上述代码是这个章节我们一直使用的代码流程, 如下是其完整的逻辑执行图

如果放在集群中运行, 通过 WebUI 可以查看到如下 DAG 结构

Step 1: 从 ResultStage 开始执行

最接近 Result 部分的 Stage id 为 0, 这个 Stage 被称之为 ResultStage

由代码可以知道, 最终调用 Action 促使整个流程执行的是最后一个 RDD, strRDD.collect, 所以当执行 RDD 的计算时候, 先计算的也是这个 RDD

Step 2: RDD 之间是有关联的

前面已经知道, 最后一个 RDD 先得到执行机会, 先从这个 RDD 开始执行, 但是这个 RDD 中有数据吗 ? 如果没有数据, 它的计算是什么? 它的计算是从父 RDD 中获取数据, 并执行传入的算子的函数

简单来说, 从产生 Result 的地方开始计算, 但是其 RDD 中是没数据的, 所以会找到父 RDD 来要数据, 父 RDD 也没有数据, 继续向上要, 所以, 计算从 Result 处调用, 但是从整个逻辑图中的最左边 RDD 开始, 类似一个递归的过程

6.3. 调度过程

导读

生成逻辑图和物理图的系统组件

Job 和 Stage, Task 之间的关系

如何调度 Job

逻辑图

是什么 怎么生成 具体怎么生成

1
2
3
4
5

`val textRDD = sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD = textRDD.flatMap(_.split(" "))
val tupleRDD = splitRDD.map((_, 1))
val reduceRDD = tupleRDD.reduceByKey(_ + _)
val strRDD = reduceRDD.map(item => s"${item._1}, ${item._2}")`

逻辑图如何生成

上述代码在 Spark Application 的 main 方法中执行, 而 Spark Application 在 Driver 中执行, 所以上述代码在 Driver 中被执行, 那么这段代码执行的结果是什么呢?

一段 Scala 代码的执行结果就是最后一行的执行结果, 所以上述的代码, 从逻辑上执行结果就是最后一个 RDD, 最后一个 RDD 也可以认为就是逻辑执行图, 为什么呢?

例如 rdd2 = rdd1.map(…) 中, 其实本质上 rdd2 是一个类型为 MapPartitionsRDD 的对象, 而创建这个对象的时候, 会通过构造函数传入当前 RDD 对象, 也就是父 RDD, 也就是调用 map 算子的 rdd1, rdd1 是 rdd2 的父 RDD

一个 RDD 依赖另外一个 RDD, 这个 RDD 又依赖另外的 RDD, 一个 RDD 可以通过 getDependency 获得其父 RDD, 这种环环相扣的关系, 最终从最后一个 RDD 就可以推演出前面所有的 RDD

逻辑图是什么, 干啥用

逻辑图其实本质上描述的就是数据的计算过程, 数据从哪来, 经过什么样的计算, 得到什么样的结果, 再执行什么计算, 得到什么结果

可是数据的计算是描述好了, 这种计算该如何执行呢?

物理图

数据的计算表示好了, 该正式执行了, 但是如何执行? 如何执行更快更好更酷? 就需要为其执行做一个规划, 所以需要生成物理执行图

1

`strRDD.collect.foreach(item => println(item))`

上述代码其实就是最后的一个 RDD 调用了 Action 方法, 调用 Action 方法的时候, 会请求一个叫做 DAGScheduler 的组件, DAGScheduler 会创建用于执行 RDD 的 Stage 和 Task

DAGScheduler 是一个由 SparkContext 创建, 运行在 Driver 上的组件, 其作用就是将由 RDD 构建出来的逻辑计划, 构建成为由真正在集群中运行的 Task 组成的物理执行计划, DAGScheduler 主要做如下三件事

帮助每个 Job 计算 DAG 并发给 TaskSheduler 调度

确定每个 Task 的最佳位置

跟踪 RDD 的缓存状态, 避免重新计算

从字面意思上来看, DAGScheduler 是调度 DAG 去运行的, DAG 被称作为有向无环图, 其实可以将 DAG 理解为就是 RDD 的逻辑图, 其呈现两个特点: RDD 的计算是有方向的, RDD 的计算是无环的, 所以 DAGScheduler 也可以称之为 RDD Scheduler, 但是真正运行在集群中的并不是 RDD, 而是 Task 和 Stage, DAGScheduler 负责这种转换

Job 是什么 ?

Job 什么时候生成 ?

当一个 RDD 调用了 Action 算子的时候, 在 Action 算子内部, 会使用 sc.runJob() 调用 SparkContext 中的 runJob 方法, 这个方法又会调用 DAGScheduler 中的 runJob, 后在 DAGScheduler 中使用消息驱动的形式创建 Job

简而言之, Job 在 RDD 调用 Action 算子的时候生成, 而且调用一次 Action 算子, 就会生成一个 Job, 如果一个 SparkApplication 中调用了多次 Action 算子, 会生成多个 Job 串行执行, 每个 Job 独立运作, 被独立调度, 所以 RDD 的计算也会被执行多次

Job 是什么 ?

如果要将 Spark 的程序调度到集群中运行, Job 是粒度最大的单位, 调度以 Job 为最大单位, 将 Job 拆分为 Stage 和 Task 去调度分发和运行, 一个 Job 就是一个 Spark 程序从 读取 → 计算 → 运行 的过程

一个 Spark Application 可以包含多个 Job, 这些 Job 之间是串行的, 也就是第二个 Job 需要等待第一个 Job 的执行结束后才会开始执行

Job 和 Stage 的关系

Job 是一个最大的调度单位, 也就是说 DAGScheduler 会首先创建一个 Job 的相关信息, 后去调度 Job, 但是没办法直接调度 Job, 比如说现在要做一盘手撕包菜, 不可能直接去炒一整颗包菜, 要切好撕碎, 再去炒

为什么 Job 需要切分 ?

• 因为 Job 的含义是对整个 RDD 血统求值, 但是 RDD 之间可能会有一些宽依赖

• 如果遇到宽依赖的话, 两个 RDD 之间需要进行数据拉取和复制

  如果要进行拉取和复制的话, 那么一个 RDD 就必须等待它所依赖的 RDD 所有分区先计算完成, 然后再进行拉取

• 由上得知, 一个 Job 是无法计算完整个 RDD 血统的

如何切分 ?

创建一个 Stage, 从后向前回溯 RDD, 遇到 Shuffle 依赖就结束 Stage, 后创建新的 Stage 继续回溯. 这个过程上面已经详细的讲解过, 但是问题是切分以后如何执行呢, 从后向前还是从前向后, 是串行执行多个 Stage, 还是并行执行多个 Stage

问题一: 执行顺序

在图中, Stage 0 的计算需要依赖 Stage 1 的数据, 因为 reduceRDD 中一个分区可能需要多个 tupleRDD 分区的数据, 所以 tupleRDD 必须先计算完, 所以, 应该在逻辑图中自左向右执行 Stage

问题二: 串行还是并行

还是同样的原因, Stage 0 如果想计算, Stage 1 必须先计算完, 因为 Stage 0 中每个分区都依赖 Stage 1 中的所有分区, 所以 Stage 1 不仅需要先执行, 而且 Stage 1 执行完之前 Stage 0 无法执行, 它们只能串行执行

总结

一个 Stage 就是物理执行计划中的一个步骤, 一个 Spark Job 就是划分到不同 Stage 的计算过程

Stage 之间的边界由 Shuffle 操作来确定

1

*   `Stage` 内的 `RDD` 之间都是窄依赖, 可以放在一个管道中执行

• 而 Shuffle 后的 Stage 需要等待前面 Stage 的执行

Stage 有两种

1

ShuffMapStage`, 其中存放窄依赖的 `RDD

ResultStage, 每个 Job 只有一个, 负责计算结果, 一个 ResultStage 执行完成标志着整个 Job 执行完毕

Stage 和 Task 的关系

前面我们说到 Job 无法直接执行, 需要先划分为多个 Stage, 去执行 Stage, 那么 Stage 可以直接执行吗?

第一点: Stage 中的 RDD 之间是窄依赖

因为 Stage 中的所有 RDD 之间都是窄依赖, 窄依赖 RDD 理论上是可以放在同一个 Pipeline(管道, 流水线) 中执行的, 似乎可以直接调度 Stage 了? 其实不行, 看第二点

第二点: 别忘了 RDD 还有分区

一个 RDD 只是一个概念, 而真正存放和处理数据时, 都是以分区作为单位的

Stage 对应的是多个整体上的 RDD, 而真正的运行是需要针对 RDD 的分区来进行的

第三点: 一个 Task 对应一个 RDD 的分区

一个比 Stage 粒度更细的单元叫做 Task, Stage 是由 Task 组成的, 之所以有 Task 这个概念, 是因为 Stage 针对整个 RDD, 而计算的时候, 要针对 RDD 的分区

假设一个 Stage 中有 10 个 RDD, 这些 RDD 中的分区各不相同, 但是分区最多的 RDD 有 30 个分区, 而且很显然, 它们之间是窄依赖关系

那么, 这个 Stage 中应该有多少 Task 呢? 应该有 30 个 Task, 因为一个 Task 计算一个 RDD 的分区. 这个 Stage 至多有 30 个分区需要计算

总结

1

*   一个 `Stage` 就是一组并行的 `Task` 集合

• Task 是 Spark 中最小的独立执行单元, 其作用是处理一个 RDD 分区
• 一个 Task 只可能存在于一个 Stage 中, 并且只能计算一个 RDD 的分区

TaskSet

梳理一下这几个概念, Job > Stage > Task, Job 中包含 Stage 中包含 Task

而 Stage 中经常会有一组 Task 需要同时执行, 所以针对于每一个 Task 来进行调度太过繁琐, 而且没有意义, 所以每个 Stage 中的 Task 们会被收集起来, 放入一个 TaskSet 集合中

一个 Stage 有一个 TaskSet

TaskSet 中 Task 的个数由 Stage 中的最大分区数决定

整体执行流程

6.3. Shuffle 过程

导读

本章节重点是介绍 Shuffle 的流程, 因为根据 ShuffleWriter 的实现不同, 其过程也不同, 所以前半部分根据默认的存储引擎 SortShuffleWriter 来讲解

后半部分简要介绍一下其它的 ShuffleWriter

Shuffle 过程的组件结构

从整体视角上来看, Shuffle 发生在两个 Stage 之间, 一个 Stage 把数据计算好, 整理好, 等待另外一个 Stage 来拉取

放大视角, 会发现, 其实 Shuffle 发生在 Task 之间, 一个 Task 把数据整理好, 等待 Reducer 端的 Task 来拉取

如果更细化一下, Task 之间如何进行数据拷贝的呢? 其实就是一方 Task 把文件生成好, 然后另一方 Task 来拉取

现在是一个 Reducer 的情况, 如果有多个 Reducer 呢? 如果有多个 Reducer 的话, 就可以在每个 Mapper 为所有的 Reducer 生成各一个文件, 这种叫做 Hash base shuffle, 这种 Shuffle 的方式问题大家也知道, 就是生成中间文件过多, 而且生成文件的话需要缓冲区, 占用内存过大

那么可以把这些文件合并起来, 生成一个文件返回, 这种 Shuffle 方式叫做 Sort base shuffle, 每个 Reducer 去文件的不同位置拿取数据

如果再细化一下, 把参与这件事的组件也放置进去, 就会是如下这样

有哪些 ShuffleWriter ?

大致上有三个 ShufflWriter, Spark 会按照一定的规则去使用这三种不同的 Writer

1

BypassMergeSortShuffleWriter

这种 Shuffle Writer 也依然有 Hash base shuffle 的问题, 它会在每一个 Mapper 端对所有的 Reducer 生成一个文件, 然后再合并这个文件生成一个统一的输出文件, 这个过程中依然是有很多文件产生的, 所以只适合在小量数据的场景下使用

Spark 有考虑去掉这种 Writer, 但是因为结构中有一些依赖, 所以一直没去掉

当 Reducer 个数小于 spark.shuffle.sort.bypassMergeThreshold, 并且没有 Mapper 端聚合的时候启用这种方式

1

SortShuffleWriter

这种 ShuffleWriter 写文件的方式非常像 MapReduce 了, 后面详说

当其它两种 Shuffle 不符合开启条件时, 这种 Shuffle 方式是默认的

1

UnsafeShuffleWriter

这种 ShuffWriter 会将数据序列化, 然后放入缓冲区进行排序, 排序结束后 Spill 到磁盘, 最终合并 Spill 文件为一个大文件, 同时在进行内存存储的时候使用了 Java 得 Unsafe API, 也就是使用堆外内存, 是钨丝计划的一部分

也不是很常用, 只有在满足如下三个条件时候才会启用

1

*   序列化器序列化后的数据, 必须支持排序

• 没有 Mapper 端的聚合
• Reducer 的个数不能超过支持的上限 (2 ^ 24)

SortShuffleWriter 的执行过程

整个 SortShuffleWriter 如上述所说, 大致有如下几步

1. 首先 SortShuffleWriter 在 write 方法中回去写文件, 这个方法中创建了 ExternalSorter

2. write 中将数据 insertAll 到 ExternalSorter 中

3. 在 ExternalSorter 中排序

   1	1. 如果要聚合, 放入 `AppendOnlyMap` 中, 如果不聚合, 放入 `PartitionedPairBuffer` 中

   1. 在数据结构中进行排序, 排序过程中如果内存数据大于阈值则溢写到磁盘

使用 ExternalSorter 的 writePartitionedFile 写入输入文件

1

1.  将所有的溢写文件通过类似 `MergeSort` 的算法合并

1. 将数据写入最终的目标文件中

7. RDD 的分布式共享变量

目标

理解闭包以及 Spark 分布式运行代码的根本原理

理解累加变量的使用场景

理解广播的使用场景

什么是闭包

闭包是一个必须要理解, 但是又不太好理解的知识点, 先看一个小例子

1
2
3
4
5
6
7
8
9
10
11
12

`@Test
def test(): Unit = {
  val areaFunction = closure()
  val area = areaFunction(2)
  println(area)
}

def closure(): Int => Double = {
val factor = 3.14
val areaFunction = (r: Int) => math.pow(r, 2) * factor
areaFunction
}`

上述例子中, closure方法返回的一个函数的引用, 其实就是一个闭包, 闭包本质上就是一个封闭的作用域, 要理解闭包, 是一定要和作用域联系起来的.

能否在 test 方法中访问 closure 定义的变量?

1
2
3
4
5
6
7
8

`@Test
def test(): Unit = {
  println(factor)
}

def closure(): Int => Double = {
val factor = 3.14
}`

有没有什么间接的方式?

1
2
3
4
5
6
7
8
9
10
11

`@Test
def test(): Unit = {
  val areaFunction = closure()
  areaFunction()
}

def closure(): () => Unit = {
val factor = 3.14
val areaFunction = () => println(factor)
areaFunction
}`

什么是闭包?

1
2

`val areaFunction = closure()
areaFunction()`

通过 closure 返回的函数 areaFunction 就是一个闭包, 其函数内部的作用域并不是 test 函数的作用域, 这种连带作用域一起打包的方式, 我们称之为闭包, 在 Scala 中

• Scala 中的闭包本质上就是一个对象, 是 FunctionX 的实例*

分发闭包

1
2
3

`sc.textFile("dataset/access_log_sample.txt")
  .flatMap(item => item.split(""))
  .collect()`

上述这段代码中, flatMap 中传入的是另外一个函数, 传入的这个函数就是一个闭包, 这个闭包会被序列化运行在不同的 Executor 中

1
2
3
4
5
6
7

`class MyClass {
  val field = "Hello"

def doStuff(rdd: RDD[String]): RDD[String] = {
rdd.map(x => field + x)
}
}`

这段代码中的闭包就有了一个依赖, 依赖于外部的一个类, 因为传递给算子的函数最终要在 Executor 中运行, 所以需要 序列化 MyClass 发给每一个 Executor, 从而在 Executor 访问 MyClass 对象的属性

总结

1. 闭包就是一个封闭的作用域, 也是一个对象
2. Spark 算子所接受的函数, 本质上是一个闭包, 因为其需要封闭作用域, 并且序列化自身和依赖, 分发到不同的节点中运行

7.1. 累加器

一个小问题

1
2
3
4
5
6
7

`var count = 0

val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
sc.parallelize(Seq(1, 2, 3, 4, 5))
.foreach(count += _)
println(count)`

上面这段代码是一个非常错误的使用, 请不要仿照, 这段代码只是为了证明一些事情

先明确两件事, var count = 0 是在 Driver 中定义的, foreach(count += _) 这个算子以及传递进去的闭包运行在 Executor 中

这段代码整体想做的事情是累加一个变量, 但是这段代码的写法却做不到这件事, 原因也很简单, 因为具体的算子是闭包, 被分发给不同的节点运行, 所以这个闭包中累加的并不是 Driver 中的这个变量

全局累加器

Accumulators(累加器) 是一个只支持 added(添加) 的分布式变量, 可以在分布式环境下保持一致性, 并且能够做到高效的并发.

原生 Spark 支持数值型的累加器, 可以用于实现计数或者求和, 开发者也可以使用自定义累加器以实现更高级的需求

1
2
3
4
5
6
7
8

`val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)

val counter = sc.longAccumulator("counter")
sc.parallelize(Seq(1, 2, 3, 4, 5))
.foreach(counter.add(_))
// 运行结果: 15
println(counter.value)`

注意点:

• Accumulator 是支持并发并行的, 在任何地方都可以通过 add 来修改数值, 无论是 Driver 还是 Executor
• 只能在 Driver 中才能调用 value 来获取数值

在 WebUI 中关于 Job 部分也可以看到 Accumulator 的信息, 以及其运行的情况

累计器件还有两个小特性, 第一, 累加器能保证在 Spark 任务出现问题被重启的时候不会出现重复计算. 第二, 累加器只有在 Action 执行的时候才会被触发.

1
2
3
4
5
6
7
8

`val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)

val counter = sc.longAccumulator("counter")
sc.parallelize(Seq(1, 2, 3, 4, 5))
.map(counter.add(_)) // 这个地方不是 Action, 而是一个 Transformation
// 运行结果是 0
println(counter.value)`

自定义累加器

开发者可以通过自定义累加器来实现更多类型的累加器, 累加器的作用远远不只是累加, 比如可以实现一个累加器, 用于向里面添加一些运行信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

`class InfoAccumulator extends AccumulatorV2[String, Set[String]] {
  private val infos: mutable.Set[String] = mutable.Set()

override def isZero: Boolean = {
infos.isEmpty
}
override def copy(): AccumulatorV2[String, Set[String]] = {
val newAccumulator = new InfoAccumulator()
infos.synchronized {
newAccumulator.infos ++= infos
}
newAccumulator
}
override def reset(): Unit = {
infos.clear()
}
override def add(v: String): Unit = {
infos += v
}
override def merge(other: AccumulatorV2[String, Set[String]]): Unit = {
infos ++= other.value
}
override def value: Set[String] = {
infos.toSet
}
}
@Test
def accumulator2(): Unit = {
val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)
val infoAccumulator = new InfoAccumulator()
sc.register(infoAccumulator, "infos")
sc.parallelize(Seq("1", "2", "3"))
.foreach(item => infoAccumulator.add(item))
// 运行结果: Set(3, 1, 2)
println(infoAccumulator.value)
sc.stop()
}`

注意点:

可以通过继承 AccumulatorV2 来创建新的累加器

有几个方法需要重写

1

*   reset 方法用于把累加器重置为 0

• add 方法用于把其它值添加到累加器中
• merge 方法用于指定如何合并其他的累加器

value 需要返回一个不可变的集合, 因为不能因为外部的修改而影响自身的值

7.2. 广播变量

目标

1. 理解为什么需要广播变量, 以及其应用场景
2. 能够通过代码使用广播变量

广播变量的作用

广播变量允许开发者将一个 Read-Only 的变量缓存到集群中每个节点中, 而不是传递给每一个 Task 一个副本.

• 集群中每个节点, 指的是一个机器
• 每一个 Task, 一个 Task 是一个 Stage 中的最小处理单元, 一个 Executor 中可以有多个 Stage, 每个 Stage 有多个 Task

所以在需要跨多个 Stage 的多个 Task 中使用相同数据的情况下, 广播特别的有用

广播变量的API

使用广播变量的一般套路

可以通过如下方式创建广播变量

1

`val b = sc.broadcast(1)`

如果 Log 级别为 DEBUG 的时候, 会打印如下信息

1
2
3
4
5

`DEBUG BlockManager: Put block broadcast_0 locally took  430 ms
DEBUG BlockManager: Putting block broadcast_0 without replication took  431 ms
DEBUG BlockManager: Told master about block broadcast_0_piece0
DEBUG BlockManager: Put block broadcast_0_piece0 locally took  4 ms
DEBUG BlockManager: Putting block broadcast_0_piece0 without replication took  4 ms`

创建后可以使用 value 获取数据

1

`b.value`

获取数据的时候会打印如下信息

1
2

`DEBUG BlockManager: Getting local block broadcast_0
DEBUG BlockManager: Level for block broadcast_0 is StorageLevel(disk, memory, deserialized, 1 replicas)`

广播变量使用完了以后, 可以使用 unpersist 删除数据

1

`b.unpersist`

删除数据以后, 可以使用 destroy 销毁变量, 释放内存空间

1

`b.destroy`

销毁以后, 会打印如下信息

1
2
3
4

`DEBUG BlockManager: Removing broadcast 0
DEBUG BlockManager: Removing block broadcast_0_piece0
DEBUG BlockManager: Told master about block broadcast_0_piece0
DEBUG BlockManager: Removing block broadcast_0`

使用 value 方法的注意点

方法签名 value: T

在 value 方法内部会确保使用获取数据的时候, 变量必须是可用状态, 所以必须在变量被 destroy 之前使用 value 方法, 如果使用 value 时变量已经失效, 则会爆出以下错误

1
2
3
4

`org.apache.spark.SparkException: Attempted to use Broadcast(0) after it was destroyed (destroy at <console>:27)
  at org.apache.spark.broadcast.Broadcast.assertValid(Broadcast.scala:144)
  at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:69)
  ... 48 elided`

使用 destroy 方法的注意点

方法签名 destroy(): Unit

destroy 方法会移除广播变量, 彻底销毁掉, 但是如果你试图多次 destroy 广播变量, 则会爆出以下错误

1
2
3
4
5

`org.apache.spark.SparkException: Attempted to use Broadcast(0) after it was destroyed (destroy at <console>:27)
  at org.apache.spark.broadcast.Broadcast.assertValid(Broadcast.scala:144)
  at org.apache.spark.broadcast.Broadcast.destroy(Broadcast.scala:107)
  at org.apache.spark.broadcast.Broadcast.destroy(Broadcast.scala:98)
  ... 48 elided`

广播变量的使用场景

假设我们在某个算子中需要使用一个保存了项目和项目的网址关系的 Map[String, String] 静态集合, 如下

1
2
3

`val pws = Map("Apache Spark" -> "http://spark.apache.org/", "Scala" -> "http://www.scala-lang.org/")

val websites = sc.parallelize(Seq("Apache Spark", "Scala")).map(pws).collect`

上面这段代码是没有问题的, 可以正常运行的, 但是非常的低效, 因为虽然可能 pws 已经存在于某个 Executor 中了, 但是在需要的时候还是会继续发往这个 Executor, 如果想要优化这段代码, 则需要尽可能的降低网络开销

可以使用广播变量进行优化, 因为广播变量会缓存在集群中的机器中, 比 Executor 在逻辑上更 “大”

1
2

`val pwsB = sc.broadcast(pws)
val websites = sc.parallelize(Seq("Apache Spark", "Scala")).map(pwsB.value).collect`

上面两段代码所做的事情其实是一样的, 但是当需要运行多个 Executor (以及多个 Task) 的时候, 后者的效率更高

扩展

正常情况下使用 Task 拉取数据的时候, 会将数据拷贝到 Executor 中多次, 但是使用广播变量的时候只会复制一份数据到 Executor 中, 所以在两种情况下特别适合使用广播变量

一个 Executor 中有多个 Task 的时候

一个变量比较大的时候

而且在 Spark 中还有一个约定俗称的做法, 当一个 RDD 很大并且还需要和另外一个 RDD 执行 join 的时候, 可以将较小的 RDD 广播出去, 然后使用大的 RDD 在算子 map 中直接 join, 从而实现在 Map 端 join

1
2
3
4
5
6

`val acMap = sc.broadcast(myRDD.map { case (a,b,c,b) => (a, c) }.collectAsMap)
val otherMap = sc.broadcast(myOtherRDD.collectAsMap)

myBigRDD.map { case (a, b, c, d) =>
(acMap.value.get(a).get, otherMap.value.get(c).get)
}.collect

一般情况下在这种场景下, 会广播 Map 类型的数据, 而不是数组, 因为这样容易使用 Key 找到对应的 Value 简化使用

总结

1. 广播变量用于将变量缓存在集群中的机器中, 避免机器内的 Executors 多次使用网络拉取数据
2. 广播变量的使用步骤: (1) 创建 (2) 在 Task 中获取值 (3) 销毁

SparkSQL

目标

1. SparkSQL 是什么
2. SparkSQL 如何使用

1. SparkSQL 是什么

目标

对于一件事的理解, 应该分为两个大部分, 第一, 它是什么, 第二, 它解决了什么问题

1. 理解为什么会有 SparkSQL
2. 理解 SparkSQL 所解决的问题, 以及它的使命

1.1. SparkSQL 的出现契机

目标

理解 SparkSQL 是什么

主线

1. 历史前提
2. 发展过程
3. 重要性

数据分析的方式

数据分析的方式大致上可以划分为 SQL 和 命令式两种

命令式

在前面的 RDD 部分, 非常明显可以感觉的到是命令式的, 主要特征是通过一个算子, 可以得到一个结果, 通过结果再进行后续计算.

1
2
3
4
5

sc.textFile("...")
  .flatMap(_.split(" "))
  .map((_, 1))
  .reduceByKey(_ + _)
  .collect()

命令式的优点

• 操作粒度更细, 能够控制数据的每一个处理环节
• 操作更明确, 步骤更清晰, 容易维护
• 支持非结构化数据的操作

命令式的缺点

• 需要一定的代码功底
• 写起来比较麻烦

SQL

对于一些数据科学家, 要求他们为了做一个非常简单的查询, 写一大堆代码, 明显是一件非常残忍的事情, 所以 SQL on Hadoop 是一个非常重要的方向.

1
2
3
4
5
6

SELECT
    name,
    age,
    school
FROM students
WHERE age > 10

SQL 的优点

• 表达非常清晰, 比如说这段 SQL 明显就是为了查询三个字段, 又比如说这段 SQL 明显能看到是想查询年龄大于 10 岁的条目

SQL 的缺点

• 想想一下 3 层嵌套的 SQL, 维护起来应该挺力不从心的吧
• 试想一下, 如果使用 SQL 来实现机器学习算法, 也挺为难的吧

SQL 擅长数据分析和通过简单的语法表示查询, 命令式操作适合过程式处理和算法性的处理. 在 Spark 出现之前, 对于结构化数据的查询和处理, 一个工具一向只能支持 SQL 或者命令式, 使用者被迫要使用多个工具来适应两种场景, 并且多个工具配合起来比较费劲.

而 Spark 出现了以后, 统一了两种数据处理范式, 是一种革新性的进步.

因为 SQL 是数据分析领域一个非常重要的范式, 所以 Spark 一直想要支持这种范式, 而伴随着一些决策失误, 这个过程其实还是非常曲折的

Hive

解决的问题

• Hive 实现了 SQL on Hadoop, 使用 MapReduce 执行任务
• 简化了 MapReduce 任务

新的问题

• Hive 的查询延迟比较高, 原因是使用 MapReduce 做调度

Shark

解决的问题

• Shark 改写 Hive 的物理执行计划, 使用 Spark 作业代替 MapReduce 执行物理计划
• 使用列式内存存储
• 以上两点使得 Shark 的查询效率很高

新的问题

• Shark 重用了 Hive 的 SQL 解析, 逻辑计划生成以及优化, 所以其实可以认为 Shark 只是把 Hive 的物理执行替换为了 Spark 作业
• 执行计划的生成严重依赖 Hive, 想要增加新的优化非常困难
• Hive 使用 MapReduce 执行作业, 所以 Hive 是进程级别的并行, 而 Spark 是线程级别的并行, 所以 Hive 中很多线程不安全的代码不适用于 Spark

由于以上问题, Shark 维护了 Hive 的一个分支, 并且无法合并进主线, 难以为继

1

SparkSQL

解决的问题

• Spark SQL 使用 Hive 解析 SQL 生成 AST 语法树, 将其后的逻辑计划生成, 优化, 物理计划都自己完成, 而不依赖 Hive
• 执行计划和优化交给优化器 Catalyst
• 内建了一套简单的 SQL 解析器, 可以不使用 HQL, 此外, 还引入和 DataFrame 这样的 DSL API, 完全可以不依赖任何 Hive 的组件
• Shark 只能查询文件, Spark SQL 可以直接降查询作用于 RDD, 这一点是一个大进步

新的问题

对于初期版本的 SparkSQL, 依然有挺多问题, 例如只能支持 SQL 的使用, 不能很好的兼容命令式, 入口不够统一等

1

Dataset

SparkSQL 在 2.0 时代, 增加了一个新的 API, 叫做 Dataset, Dataset 统一和结合了 SQL 的访问和命令式 API 的使用, 这是一个划时代的进步

在 Dataset 中可以轻易的做到使用 SQL 查询并且筛选数据, 然后使用命令式 API 进行探索式分析

	重要性SparkSQL 不只是一个 SQL 引擎, SparkSQL 也包含了一套对 结构化数据的命令式 API, 事实上, 所有 Spark 中常见的工具, 都是依赖和依照于 SparkSQL 的 API 设计的

总结: SparkSQL 是什么

1

SparkSQL` 是一个为了支持 `SQL` 而设计的工具, 但同时也支持命令式的 `API

1.2. SparkSQL 的适用场景

目标

理解 SparkSQL 的适用场景

结构化数据

一般指数据有固定的 Schema, 例如在用户表中, name 字段是 String 型, 那么每一条数据的 name 字段值都可以当作 String 来使用

1
2
3
4
5
6
7
8
9

+----+--------------+---------------------------+-------+---------+
| id | name         | url                       | alexa | country |
+----+--------------+---------------------------+-------+---------+
| 1  | Google       | https://www.google.cm/    | 1     | USA     |
| 2  | 淘宝          | https://www.taobao.com/   | 13    | CN      |
| 3  | 菜鸟教程      | http://www.runoob.com/    | 4689  | CN      |
| 4  | 微博          | http://weibo.com/         | 20    | CN      |
| 5  | Facebook     | https://www.facebook.com/ | 3     | USA     |
+----+--------------+---------------------------+-------+---------+

半结构化数据

一般指的是数据没有固定的 Schema, 但是数据本身是有结构的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

{
     "firstName": "John",
     "lastName": "Smith",
     "age": 25,
     "phoneNumber":
     [
         {
           "type": "home",
           "number": "212 555-1234"
         },
         {
           "type": "fax",
           "number": "646 555-4567"
         }
     ]
 }

没有固定 Schema

指的是半结构化数据是没有固定的 Schema 的, 可以理解为没有显式指定 Schema
比如说一个用户信息的 JSON 文件, 第一条数据的 phone_num 有可能是 String, 第二条数据虽说应该也是 String, 但是如果硬要指定为 BigInt, 也是有可能的
因为没有指定 Schema, 没有显式的强制的约束

有结构

虽说半结构化数据是没有显式指定 Schema 的, 也没有约束, 但是半结构化数据本身是有有隐式的结构的, 也就是数据自身可以描述自身
例如 JSON 文件, 其中的某一条数据是有字段这个概念的, 每个字段也有类型的概念, 所以说 JSON 是可以描述自身的, 也就是数据本身携带有元信息

SparkSQL 处理什么数据的问题?

• Spark 的 RDD 主要用于处理 非结构化数据 和 半结构化数据
• SparkSQL 主要用于处理 结构化数据

SparkSQL 相较于 RDD 的优势在哪?

• SparkSQL 提供了更好的外部数据源读写支持
  • 因为大部分外部数据源是有结构化的, 需要在 RDD 之外有一个新的解决方案, 来整合这些结构化数据源
• SparkSQL 提供了直接访问列的能力
  • 因为 SparkSQL 主要用做于处理结构化数据, 所以其提供的 API 具有一些普通数据库的能力

总结: SparkSQL 适用于什么场景?

SparkSQL 适用于处理结构化数据的场景

本章总结

• SparkSQL 是一个即支持 SQL 又支持命令式数据处理的工具
• SparkSQL 的主要适用场景是处理结构化数据

2. SparkSQL 初体验

目标

1. 了解 SparkSQL 的 API 由哪些部分组成

2.3. RDD 版本的 WordCount

1
2
3
4
5
6
7
8

val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)

sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")
  .flatMap(.split(" "))
  .map((, 1))
  .reduceByKey(_ + _)
  .collect

• RDD 版本的代码有一个非常明显的特点, 就是它所处理的数据是基本类型的, 在算子中对整个数据进行处理

2.2. 命令式 API 的入门案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

case class People(name: String, age: Int)

val spark: SparkSession = new sql.SparkSession.Builder()       (1)
  .appName("hello")
  .master("local[6]")
  .getOrCreate()
import spark.implicits._
val peopleRDD: RDD[People] = spark.sparkContext.parallelize(Seq(People("zhangsan", 9), People("lisi", 15)))
val peopleDS: Dataset[People] = peopleRDD.toDS()               (2)
val teenagers: Dataset[String] = peopleDS.where('age > 10)     (3)
  .where('age < 20)
  .select('name)
  .as[String]
/*
+----+
|name|
+----+
|lisi|
+----+
*/
teenagers.show()

SparkSession

SparkContext 作为 RDD 的创建者和入口, 其主要作用有如下两点

• 创建 RDD, 主要是通过读取文件创建 RDD
• 监控和调度任务, 包含了一系列组件, 例如 DAGScheduler, TaskSheduler

为什么无法使用 SparkContext 作为 SparkSQL 的入口?

• SparkContext 在读取文件的时候, 是不包含 Schema 信息的, 因为读取出来的是 RDD
• SparkContext 在整合数据源如 Cassandra, JSON, Parquet 等的时候是不灵活的, 而 DataFrame 和 Dataset 一开始的设计目标就是要支持更多的数据源
• SparkContext 的调度方式是直接调度 RDD, 但是一般情况下针对结构化数据的访问, 会先通过优化器优化一下

所以 SparkContext 确实已经不适合作为 SparkSQL 的入口, 所以刚开始的时候 Spark 团队为 SparkSQL 设计了两个入口点, 一个是 SQLContext 对应 Spark 标准的 SQL 执行, 另外一个是 HiveContext 对应 HiveSQL 的执行和 Hive 的支持.

在 Spark 2.0 的时候, 为了解决入口点不统一的问题, 创建了一个新的入口点 SparkSession, 作为整个 Spark 生态工具的统一入口点, 包括了 SQLContext, HiveContext, SparkContext 等组件的功能

新的入口应该有什么特性?

• 能够整合 SQLContext, HiveContext, SparkContext, StreamingContext 等不同的入口点
• 为了支持更多的数据源, 应该完善读取和写入体系
• 同时对于原来的入口点也不能放弃, 要向下兼容

DataFrame & Dataset

1

SparkSQL` 最大的特点就是它针对于结构化数据设计, 所以 `SparkSQL` 应该是能支持针对某一个字段的访问的, 而这种访问方式有一个前提, 就是 `SparkSQL` 的数据集中, 要 **包含结构化信息**, 也就是俗称的 `Schema

而 SparkSQL 对外提供的 API 有两类, 一类是直接执行 SQL, 另外一类就是命令式. SparkSQL 提供的命令式 API 就是 DataFrame 和 Dataset, 暂时也可以认为 DataFrame 就是 Dataset, 只是在不同的 API 中返回的是 Dataset 的不同表现形式

1
2
3
4
5
6

// RDD
rdd.map { case Person(id, name, age) => (age, 1) }
  .reduceByKey {case ((age, count), (totalAge, totalCount)) => (age, count + totalCount)}

// DataFrame
df.groupBy("age").count("age")

通过上面的代码, 可以清晰的看到, SparkSQL 的命令式操作相比于 RDD 来说, 可以直接通过 Schema 信息来访问其中某个字段, 非常的方便

2.2. SQL 版本 WordCount

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

val spark: SparkSession = new sql.SparkSession.Builder()
  .appName("hello")
  .master("local[6]")
  .getOrCreate()

import spark.implicits._
val peopleRDD: RDD[People] = spark.sparkContext.parallelize(Seq(People("zhangsan", 9), People("lisi", 15)))
val peopleDS: Dataset[People] = peopleRDD.toDS()
peopleDS.createOrReplaceTempView("people")
val teenagers: DataFrame = spark.sql("select name from people where age > 10 and age < 20")
/*
+----+
|name|
+----+
|lisi|
+----+
 */
teenagers.show()

以往使用 SQL 肯定是要有一个表的, 在 Spark 中, 并不存在表的概念, 但是有一个近似的概念, 叫做 DataFrame, 所以一般情况下要先通过 DataFrame 或者 Dataset 注册一张临时表, 然后使用 SQL 操作这张临时表

总结

SparkSQL 提供了 SQL 和 命令式 API 两种不同的访问结构化数据的形式, 并且它们之间可以无缝的衔接

命令式 API 由一个叫做 Dataset 的组件提供, 其还有一个变形, 叫做 DataFrame

3. [扩展] Catalyst 优化器

目标

1. 理解 SparkSQL 和以 RDD 为代表的 SparkCore 最大的区别
2. 理解优化器的运行原理和作用

3.1. RDD 和 SparkSQL 运行时的区别

RDD 的运行流程

大致运行步骤

先将 RDD 解析为由 Stage 组成的 DAG, 后将 Stage 转为 Task 直接运行

问题

任务会按照代码所示运行, 依赖开发者的优化, 开发者的会在很大程度上影响运行效率

解决办法

创建一个组件, 帮助开发者修改和优化代码, 但是这在 RDD 上是无法实现的

为什么 RDD 无法自我优化?

• RDD 没有 Schema 信息
• RDD 可以同时处理结构化和非结构化的数据

SparkSQL 提供了什么?

和 RDD 不同, SparkSQL 的 Dataset 和 SQL 并不是直接生成计划交给集群执行, 而是经过了一个叫做 Catalyst 的优化器, 这个优化器能够自动帮助开发者优化代码

也就是说, 在 SparkSQL 中, 开发者的代码即使不够优化, 也会被优化为相对较好的形式去执行

为什么 SparkSQL 提供了这种能力?

首先, SparkSQL 大部分情况用于处理结构化数据和半结构化数据, 所以 SparkSQL 可以获知数据的 Schema, 从而根据其 Schema 来进行优化

3.2. Catalyst

	为了解决过多依赖 Hive 的问题, SparkSQL 使用了一个新的 SQL 优化器替代 Hive 中的优化器, 这个优化器就是 Catalyst, 整个 SparkSQL 的架构大致如下API 层简单的说就是 Spark 会通过一些 API 接受 SQL 语句收到 SQL 语句以后, 将其交给 Catalyst, Catalyst 负责解析 SQL, 生成执行计划等Catalyst 的输出应该是 RDD 的执行计划最终交由集群运行

Step 1 : 解析 SQL, 并且生成 AST (抽象语法树)

Step 2 : 在 AST 中加入元数据信息, 做这一步主要是为了一些优化, 例如 col = col 这样的条件, 下图是一个简略图, 便于理解

• score.id → id#1#L 为 score.id 生成 id 为 1, 类型是 Long
• score.math_score → math_score#2#L 为 score.math_score 生成 id 为 2, 类型为 Long
• people.id → id#3#L 为 people.id 生成 id 为 3, 类型为 Long
• people.age → age#4#L 为 people.age 生成 id 为 4, 类型为 Long

Step 3 : 对已经加入元数据的 AST, 输入优化器, 进行优化, 从两种常见的优化开始, 简单介绍

• 谓词下推 Predicate Pushdown, 将 Filter 这种可以减小数据集的操作下推, 放在 Scan 的位置, 这样可以减少操作时候的数据量

• 列值裁剪 Column Pruning, 在谓词下推后, people 表之上的操作只用到了 id 列, 所以可以把其它列裁剪掉, 这样可以减少处理的数据量, 从而优化处理速度

• 还有其余很多优化点, 大概一共有一二百种, 随着 SparkSQL 的发展, 还会越来越多, 感兴趣的同学可以继续通过源码了解, 源码在 org.apache.spark.sql.catalyst.optimizer.Optimizer

Step 4 : 上面的过程生成的 AST 其实最终还没办法直接运行, 这个 AST 叫做 逻辑计划, 结束后, 需要生成 物理计划, 从而生成 RDD 来运行

• 在生成物理计划的时候, 会经过成本模型对整棵树再次执行优化, 选择一个更好的计划
• 在生成物理计划以后, 因为考虑到性能, 所以会使用代码生成, 在机器中运行

	可以使用 queryExecution 方法查看逻辑执行计划, 使用 explain 方法查看物理执行计划)也可以使用 Spark WebUI 进行查看

总结

SparkSQL 和 RDD 不同的主要点是在于其所操作的数据是结构化的, 提供了对数据更强的感知和分析能力, 能够对代码进行更深层的优化, 而这种能力是由一个叫做 Catalyst 的优化器所提供的

Catalyst 的主要运作原理是分为三步, 先对 SQL 或者 Dataset 的代码解析, 生成逻辑计划, 后对逻辑计划进行优化, 再生成物理计划, 最后生成代码到集群中以 RDD 的形式运行

4. Dataset 的特点

目标

1. 理解 Dataset 是什么
2. 理解 Dataset 的特性

Dataset 是什么?

1
2
3
4
5
6
7
8
9
10
11
12
13

val spark: SparkSession = new sql.SparkSession.Builder()
  .appName("hello")
  .master("local[6]")
  .getOrCreate()

import spark.implicits._
val dataset: Dataset[People] = spark.createDataset(Seq(People("zhangsan", 9), People("lisi", 15)))
// 方式1: 通过对象来处理
dataset.filter(item => item.age > 10).show()
// 方式2: 通过字段来处理
dataset.filter('age > 10).show()
// 方式3: 通过类似SQL的表达式来处理
dataset.filter("age > 10").show()

问题1: People 是什么?

People 是一个强类型的类

问题2: 这个 Dataset 中是结构化的数据吗?

非常明显是的, 因为 People 对象中有结构信息, 例如字段名和字段类型

问题3: 这个 Dataset 能够使用类似 SQL 这样声明式结构化查询语句的形式来查询吗?

当然可以, 已经演示过了

问题4: Dataset 是什么?

1

Dataset` 是一个强类型, 并且类型安全的数据容器, 并且提供了结构化查询 `API` 和类似 `RDD` 一样的命令式 `API

即使使用 Dataset 的命令式 API, 执行计划也依然会被优化

Dataset 具有 RDD 的方便, 同时也具有 DataFrame 的性能优势, 并且 Dataset 还是强类型的, 能做到类型安全.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

scala> spark.range(1).filter('id === 0).explain(true)

== Parsed Logical Plan ==
'Filter ('id = 0)
+- Range (0, 1, splits=8)
== Analyzed Logical Plan ==
id: bigint
Filter (id#51L = cast(0 as bigint))
+- Range (0, 1, splits=8)
== Optimized Logical Plan ==
Filter (id#51L = 0)
+- Range (0, 1, splits=8)
== Physical Plan ==
*Filter (id#51L = 0)
+- *Range (0, 1, splits=8)

Dataset 的底层是什么?

Dataset 最底层处理的是对象的序列化形式, 通过查看 Dataset 生成的物理执行计划, 也就是最终所处理的 RDD, 就可以判定 Dataset 底层处理的是什么形式的数据

1
2
3

val dataset: Dataset[People] = spark.createDataset(Seq(People("zhangsan", 9), People("lisi", 15)))
val internalRDD: RDD[InternalRow] = dataset.queryExecution.toRdd
dataset.queryExecution.toRdd` 这个 `API` 可以看到 `Dataset` 底层执行的 `RDD`, 这个 `RDD` 中的范型是 `InternalRow`, `InternalRow` 又称之为 `Catalyst Row`, 是 `Dataset` 底层的数据结构, 也就是说, 无论 `Dataset` 的范型是什么, 无论是 `Dataset[Person]` 还是其它的, 其最底层进行处理的数据结构都是 `InternalRow

所以, Dataset 的范型对象在执行之前, 需要通过 Encoder 转换为 InternalRow, 在输入之前, 需要把 InternalRow 通过 Decoder 转换为范型对象