【1-3章】Spark编程基础(Python版)_林子雨spark编程基础python版课后答案

课程资源：（林子雨）Spark编程基础(Python版)_哔哩哔哩_bilibili

---

第1章  大数据技术概述（8节）

第三次信息化浪潮：以物联网、云计算、大数据为标志

（一）大数据

大数据时代到来的原因：

• 技术支撑：存储设备（价格下降）、CPU计算能力（多核CPU）、网络带宽（单机不能够完成海量数据的存储和处理，借助网络分布式的集群运算）
• 数据产生方式的变革：运营式系统阶段（如超市购物在数据库系统中生成购物信息） —> 用户原创内容阶段 —> 感知式系统阶段（物联网感知终端，如传感器、摄像头、RFID）

大数据4V特性：

• 大量化Volume：数据量大（摩尔定律：人类在最近两年产生的数据量相当于之前产生的全部数据量）
• 快速化Velocity：处理速度快（一秒定律：数据从生成到决策响应仅需1s，1s的响应才能够具备它相应的商业价值）
• 多样化Variety：数据类型繁多，非结构化数据（存储于非关系型数据库中）& 结构化数据（具有规范行和列，存储于关系型数据库中）
• 价值密度低Value：但单点价值高

大数据影响：

• 对科学研究的影响：科学研究的四种范式分别为 实验-理论-计算（事先知道问题）-数据（从数据中发现问题）
• 对思维方式的影响：全样而非抽样（存储和算力均提升了）、效率而非精确（全样分析不存在误差放大问题，故不再苛求精确度）、相关而非因果

大数据关键技术：

• 数据采集
• 数据存储与管理（核心一）：分布式存储（解决数据存储问题），如GFS/HDFS、BigTable/HBase、NoSQL（键值、列族、图形、文档数据库）、NewSQL（如SQL Azure）
• 数据处理与分析（核心二）：分布式处理（解决数据高效计算问题），如MapReduce、Spark、Flink
• 数据隐私与安全

大数据计算模式：

• 企业不同应用场景对应不同计算模式
• 典型计算模式：
  • 批处理计算（大规模数据的批量处理，MapReduce/Spark）
  • 流计算（流数据的实时计算，实时处理实时响应，秒级/毫秒级，Storm/Flume）
  • 图计算（大规模图结构数据的处理，如地理信息系统、社交网络数据，Pregel/GraphX）
  • 查询分析计算（大规模数据的存储管理和查询分析，Dremel/Hive/Cassandra）

（二）代表性大数据技术

Hadoop、Spark、Flink、Beam 

1、Hadoop

不是单一软件，是一个生态系统

• HDFS：分布式文件系统，非结构化数据存储【Hadoop关键技术之一】
• YARN：资源调度和管理框架，分配计算所需的内存和CPU资源
• MapReduce：分布式计算框架【Hadoop关键技术之二】
• Hive：数据仓库。本身并不存储数据，数据存储在HDFS里；本质是一个编程接口，提供SQL查询分析（查询时写的是SQL语句，将SQL语句转成MapReduce程序，对底层数据进行查询分析）
• Pig：数据流处理。数据清洗、转换、加载（Pig Latin语言），一般和Hive组合使用
• Mahout：数据挖掘和机器学习算法库。实现常用的数据挖掘算法，如分类、聚类、回归等（单机版—>分布式）。用MapReduce实现的算法库，只需调接口、传参数，减少工作量
• Ambari：自动安装、部署、配置、管理Hadoop集群
• ZooKeeper：分布式协作服务，负责分布式协调一致性。如协调共享加锁、选管家等
• Hbase：分布式数据库。HDFS面向批处理，HBase面向实时计算
• Flume：日志采集工具
• Sqoop：ETL（抽取Extract，转换Transform，加载Load），将历史保存在关系型数据库中的数据抽取出来，保存到HDFS中，反之亦可。完成Hadoop系统组件之间的互通，即Hadoop与关系型数据库数据之间的导入导出

（1）MapReduce 

• 编程容易，屏蔽底层分布式并行编程细节（写MR程序跟写单机程序差别不大，自动分发任务到不同机器，并收集结果）
• 核心策略为分而治之，即把一个大的任务拆分成很多子任务，分发到不同机器上并行执行（只有满足分而治之的任务才能用MapReduce，如词频统计）

（2）YARN（Yet Another Resource Negotiator）

Hadoop2.0才出现。资源调度管理框架，实现“一个集群多个框架” 

• 离线批处理：MapReduce
• 实时交互式查询分析：Impala
• 流式数据实时分析：Storm
• 迭代计算：Spark 

以前为了防止资源打架，会独立部署各个计算框架（如1000台机器指定300台部署MapReduce计算框架，300台部署Spark计算框架，以此类推），但这导致开发成本高、集群资源利用率低、底层数据无法共享和无缝集成，YARN的出现解决了这一问题

2、Spark

不是单一软件，是一个生态系统

• Spark Core：完成RDD应用开发。满足企业批处理的需求
• Spark SQL：查询分析计算，分析关系数据
• Spark Streaming：流计算（Structured Streaming：结构化数据流）
• MLlib：机器学习算法库
• GraphX：编写图计算应用程序

Hadoop vs Spark，Hadoop存在以下缺点（主要是其中MapReduce的缺点）：

• 表达能力有限。MapReduce把复杂业务逻辑拆分成Map函数和Reduce函数，降低了分布式应用开发的复杂性，同时也限制了表达能力
• 磁盘IO开销大。MapReduce是基于磁盘开发的，不同阶段的衔接&中间结果的存储均涉及读写磁盘。如逻辑斯蒂回归、模拟退火算法、遗传算法等迭代算法都需要使用MapReduce进行反复迭代（读写磁盘），效率低
• 延迟高。任务分解为多个map和多个reduce，完成全部map任务才能进入reduce阶段，涉及任务之间的衔接开销，难以胜任多阶段的、比较复杂的计算任务，尤其是迭代式的计算

Spark有以下优点：

• Spark本质上计算模式也属于MapReduce，但它的操作不再局限于map和reduce，如filter过滤、groupBy分组、join连接等等，操作类型更多，表达能力更强
• Spark提供内存计算，把计算的中间结果放到内存中，高效提高迭代计算
• Spark是基于有向无环图DAG的任务调度机制（好于MapReduce的执行机制），流水线优化，使得很多数据可以一条线地执行下去，不用落磁盘进行读写，可以大大加快执行速度

Q：Spark会取代Hadoop吗？

• Hadoop有两大核心：存储框架HDFS（分布式文件系统）、分布式计算框架MapReduce
• Spark是一个单纯的计算框架，本身不具备存储能力，一般和HDFS组合使用（数据保存在HDFS中，借助于Spark计算）。Spark取代的是Hadoop里的计算框架MapReduce，而不是Hadoop

3、Flink

不是单一软件，是一个生态系统（批处理、查询分析、流计算、图计算、机器学习算法库也都有）。Flink是和Spark同一类型的计算框架

本质区别：Spark是基于RDD的批处理模型，Flink是基于一行行的流处理模型（实时性好于Spark Streaming） 

4、Beam

Google提出的统一编程接口 Beam SDK，自动翻译成其他引擎。但目前主流：Hadoop＋Spark

---

第二章  Spark的设计与运行原理（8节）

（一）Spark概述

背景：MapReduce磁盘读写、IO开销大 —＞提出Spark：基于内存的计算框架，构建大型的、低延迟的数据分析应用程序

• 三大分布式计算系统开源项目：Hadoop、Spark、Storm

1、Spark优点

• 运行速度快：基于内存的计算，数据很少落磁盘，循环数据流；DAG有向无环图执行引擎，优化执行过程，实现流水线优化
• 容易使用：支持Java、Scala、R、Python四种编程语言，其中Scala可通过Spark Shell进行交互式编程
• 通用性：不是单一组件，是一个完整的生态系统、完整的解决方案、技术软件栈
  • SQL查询：Spark SQL
  • 流式计算：Spark Streaming
  • 机器学习：Spark MLlib
  • 图算法组件：Spark GraphX
• 运行模式多样：
  • 单机／集群（本地集群／云端集群）都支持
  • 可访问多种数据源：分布式文件系统HDFS、数据库Cassandra、分布式数据库HBase、数据仓库Hive等（Spark是计算框架，本身不存储数据）

2、Spark vs Hadoop

• 主要是其中MapReduce的缺点：表达能力有限、磁盘IO开销大、延迟高
• Spark优点：
  • 操作类型更多，表达能力更强；
  • 内存计算，高效提高迭代运算（内存计算的意思是，能够不落磁盘尽量不落磁盘，而不是所有数据都在内存中运行，如shuffle必须要落磁盘）；
  • DAG有向无环图任务调度执行机制

  

（二）Spark生态系统

1、三大典型应用场景

• 批处理：MapReduce 
• 交互查询：数据仓库Impala
• 流处理：Storm

问题：（1）无法无缝共享，需要进行数据格式的转换；（2）维护成本较高；（3）资源利用不充分，无法做统一的资源管理分配

Spark一个软件栈满足不同应用场景需求，如SQL即席查询、实时流式计算、机器学习、图计算。Spark中各个组件可借助于Yarn进行统一资源调度分配管理

• 伯克利数据分析软件栈BDAS（Berkeley Data Analytics Stack）：

2、Spark的生态系统

技术软件栈，一站式服务

（三）Spark运行架构

1、基本概念

RDD：弹性分布式数据集。分布式内存的一个抽象概念（整个Spark编程最核心的数据抽象），提供了一种高度受限的共享内存模型

• 弹性：数据可大可小、分区数目动态可变化
• 分布式：分布式保存在多台机器的内存中

DAG：有向无环图。反映RDD之间的依赖关系，RDD操作会形成DAG

Executor：运行在工作节点（Worker Node，从节点）的一个进程（一个进程会派生出很多线程），负责运行具体的任务/Task

应用/Application：用户编写的Spark应用程序

任务/Task：运行在Executor进程上的工作单元（任务控制节点Driver Program）

作业/Job：一个Job包含多个RDD及作用于相应RDD上的各种操作。一个Spark应用程序提交后，就是分解成1到多个Job去完成的

阶段/Stage：作业的基本调度单位。每个Job会被分解成多组Task，每一组Task的集合叫Stage

Application > Job > Stage > Task 

2、架构设计 

• SparkContext 负责整个任务的调度、监控、执行、失败恢复、结果汇总等
• Driver Program生成SparkContext对象
• Cluster Manager接收SparkContext的资源申请，为Executor进程分配CPU、内存等资源
• Worker Node上的Executor进程派生线程执行Task 

Cluster Manager 集群资源管理器：

• 集群资源：CPU、内存、带宽
• 管理器：自带、YARN、Mesos 

执行应用时，Driver会向集群资源管理器申请资源，并启动Executor进程，向进程发送应用程序的代码和文件，应用程序会在进程上派发出线程去执行任务，执行结束后将结果返回给Driver，提交给用户/HDFS/关系型数据库等 

分布式系统的两种架构：对等架构P2P；一主多从架构（Spark、Hadoop都是这种） 

3、Spark运行基本流程

• 为Application构建基本的运行环境（Driver节点生成SparkContext对象，负责整个任务的调度、监控、执行、失败恢复、结果汇总等） 
• 运行Executor进程必须要有相关的内存、CPU资源。SparkContext向资源管理器申请资源，进行任务的分配和监控
• 集群资源管理器Cluster Manager接到申请后，为Executor进程分配CPU、内存资源，此时Worker Node上的Executor进程就可以启动了，可以派生出很多线程去执行任务
• 任务是怎么来的呢？SparkContext根据提交的代码（针对RDD的操作）生成DAG图，交给DAG Scheduler将DAG图分解成Stage，每个Stage包含很多Task
• Task任务如何分发呢？Task Scheduler会把每个阶段的任务分发给不同节点来处理（分发基本原则：计算向数据靠拢，尽量减小数据的移动开销，优先把计算分发到数据所在的节点，实现数据的本地化处理）
• 线程执行完任务后，把结果反馈给Task Scheduler，再反馈给DAG Scheduler，运行结束后写入数据并释放资源

4、RDD的设计与运行原理

（1）RDD设计原理 

MapReduce不适合处理迭代场景（如逻辑斯蒂回归、模拟退火算法、遗传算法），中间结果反复读写磁盘，磁盘IO开销太大（反复读写工作子集+序列化和反序列化开销）

• 序列化：把内存中的对象转化为可保存和传输的格式，如Java对象转化为二进制或字符串
• 反序列化：从可保存和传输的格式生成对象

RDD为了避免这些问题而出现，提供了抽象的数据结构。把具体应用逻辑表达为RDD转换，不同RDD转换之间的依赖关系即DAG图，优化实现数据的管道化（流水线化）处理，即一个操作结束后数据不需要落磁盘，马上输入给下一个操作，避免数据落地

• 一个RDD就是一个数据分布式对象的集合，本质上是一个只读的分区记录集合，可以分布式保存在很多机器上（若干分区，每个分区放在不同机器上，每个分区都是一个数据片段），分布式并行处理，高效并行计算
• RDD加载高度受限（只读）的共享内存模型，生成内存当中的数据集合，创建后就不能修改；转化过程中可以修改，即通过生成新的RDD来完成一个数据修改的目的
• RDD提供了丰富的操作类型，分为两大类：动作类型操作Action、转换类型操作Transformation。均支持粗粒度修改（一次只能针对RDD全集进行转换），不支持细粒度修改（不适合数据库对单条进行修改、不适合网页爬虫）
  • 转换操作：创建一个新的RDD，通过对已存在RDD进行转换得到，如map、filter、groupByKey
  • 动作操作：对RDD进行实际计算并返回结果，触发计算的执行，如count、collect、saveAsTextFile

高度受限的共享内存模型会不会影响表达能力？由于RDD提供的转换操作（map、filter、groupBy、join）十分丰富，可以将其组合实现很多功能。实践证明，Spark能力非常强大，虽是高度受限的共享内存模型，但不会影响表达能力。Spark提供了RDD的API，可以通过调用API实现对RDD的各种操作

（2）RDD典型执行过程

1. RDD读入外部数据源进行创建，如从底层分布式文件系统读取数据即可完成创建、生成RDD
2. RDD经过一系列转换Transformation操作，每一次都会产生不同的RDD供给下一个转换操作使用
3. 一系列转换操作后有一个动作类型操作Action，动作类型操作计算得到结果（转换类型操作不会计算得到结果）
   

惰性调用机制：对RDD的转换操作不会真正执行转换，只会记录转换轨迹，并不会真正发生计算。只有遇到第一个动作类型操作才会触发计算，执行从头到尾操作（从磁盘读取数据到输出） 

管道化/流水线优化：转换过程中数据不用落地磁盘，直接把一个操作的输出，给另一个操作作为输入，避免了不必要的读写磁盘开销，也无需保存中间结果 

MapReuce写代码时，若应用复杂则需要写入非常复杂的代码；但Spark每个操作都很简单，串联起来的操作集合可以完成非常复杂的功能

（3）Spark特性

• 高效的容错性：
  • 现有容错机制有数据复制（数据备份）、记录日志（如关系数据库，操作写入日志，操作失败可回滚等），但这两种方式开销太大
  • Spark有天然容错性，恢复数据可通过DAG图寻亲，DAG图即血缘关系图
• 中间结果持久化内存：
  • 而不是到磁盘，没有频繁写磁盘
  • 数据是在内存的多个RDD之间进行传递，避免了磁盘IO开销，同时也避免了不必要的序列化和反序列化开销 

（4）RDD运行原理（RDD之间的依赖关系）

一个RDD应用会分成多个作业，一个作业会被分成很多阶段

• 为什么要分成多个阶段？（依赖关系主要用于构建DAG有向无环图来表示Spark作业中的不同阶段）
• 以什么为依据拆分多个阶段？（看依赖关系是宽依赖还是窄依赖）

宽依赖：划分成多个阶段，包含shuffle操作（数据的重分区和混洗），即数据在网络上传输（任务的执行涉及磁盘和网络传输，性能较低）

• 父RDD的一个分区对应多个子RDD的分区
• 即一对多转换，如groupByKey、reduceByKey、join、sortByKey、distinct

窄依赖：不划分阶段，没有包含shuffle操作（计算可以在同一分区内完成），即数据不需要在网络上传输，性能较高

• 父RDD的一个分区对应一个子RDD的分区，即一对一转换，如map、filter、flatMap
• 或多个父RDD的分区对应一个子RDD的分区，如union

是否包含shuffle操作是划分宽窄依赖的依据（Spark根据每个转换操作来确定依赖的类型，如果一个操作不会触发shuffle，那么产生的子RDD和父RDD之间就是窄依赖；如果一个操作需要shuffle，那么产生的子RDD与父RDD之间就是宽依赖）

文件保存在HDFS中，进行多任务执行、分区处理。只要发生了shuffle操作，一定发生了来回交互的数据的分发。shuffle操作在网络中大规模地来回传输数据，不同节点之间互相传数据

宽依赖需要拆分成两阶段，窄依赖不用

• 窄依赖能够有利于作业优化，即进行流水线优化（中间不落磁盘，不join）
• 宽依赖不能进行流水线优化（只要发生shuffle一定会写磁盘，即落地等待）

宽窄依赖与转换类型有关。宽依赖通常与需要对多个分区进行合并或重排的转换操作相关，这些转换操作会导致Shuffle，产生宽依赖；而窄依赖通常与单个分区的转换操作相关，不需要Shuffle，产生窄依赖；

动作操作则触发实际的计算，通常不涉及依赖关系

（5）Spark优化原理（fork/join机制）

从一个RDD到另一个RDD的转换都是一个fork+一个join（fork即并行执行分区转换，结果汇总是join）

DAG有向无环图反向解析：遇到窄依赖就不断添加，形成管道化流水线处理；遇到宽依赖就断开，生成新的阶段Stage，因为要发生等待、洗牌（宽依赖生成不同阶段，窄依赖不断加入阶段） 

• 阶段Stage内部都是可以并行、流水线化处理；阶段之间都是发生了等待

（四）Spark部署方式

支持单机部署和集群部署。集群部署有以下三种：

1. StandAlone模式：使用Spark自带的集群资源管理器来管理整个CPU、内存资源调度。效率不高
2. Mesos模式：使用Mesos作为集群资源管理器。性能匹配好
3. Spark on Yarn：用的最多

Hadoop包含存储框架HDFS、HBase、计算框架MapReduce等。Spark和Hadoop并不对等，而是可能取代MapReduce。Hadoop的HDFS、HBase会继续发挥存储功能，存储数据拿给计算框架Spark来计算分析，它们共同来满足企业的相关应用场景需求

---

第3章  Spark环境搭建和使用方法（5节）

（一）安装Spark

Spark是一种分布式计算框架，存储需要Hadoop，运行在Linux系统上。Spark底层最终编译成Java字节码运行，故需要Java环境

• Spark 2.4.0 需要Java 8以上或JDK 1.8以上
• Hadoop 2.7.1

安装Hadoop教程（包含了安装Java）：Hadoop安装教程_单机/伪分布式配置_Hadoop2.6.0(2.7.1)/Ubuntu14.04(16.04)_厦大数据库实验室博客

安装Spark：Apache Spark™ - Unified Engine for large-scale data analytics

关于安装步骤我写过一篇详细的帖子，可以参考：PySpark安装及WordCount实现（基于Ubuntu）_如何原谅奋力过但无声的博客-CSDN博客 

# 解压安装包spark-2.4.0-bin-without-hadoop.tgz至路径/usr/local
# usr是unix software resource
sudo tar -zxf ~/Downloads/spark-2.4.0-bin-without-hadoop.tgz -C /usr/local/
cd /usr/local
sudo mv ./spark-2.4.0-bin-without-hadoop/ ./spark  # 更改文件夹名
sudo chown -R hadoop ./spark   # 此处hadoop为系统用户名，把spark目录权限赋予hadoop用户
 
# 配置Spark的classpath，这样Spark才能跟Hadoop挂接起来
cd /usr/local/spark
cp ./conf/spark-env.sh.template ./conf/spark-env.sh   # 拷贝配置文件
 
vim ./conf/spark-env.sh
# 编辑该配置文件，在第一行加上如下一行内容
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
 
# 保存配置文件后，就可以用Spark去访问。若需要使用HDFS中的文件，则在使用Spark前需要启动Hadoop
 
# 启动Spark Shell（Scala语言）
cd /usr/local/spark
bin/spark-shell

Spark与Hadoop如何交互？Spark部署模式：

• Local模式/单机模式：把Hadoop配置成伪分布式模式（把NameNode和DataNode都放在一台笔记本电脑上）

注：HDFS的NameNode（数据目录）和DataNode（具体存储数据），一般为一主多从架构，即一个NameNode，其余全为DataNode 

• 三种集群模式：
  • Standalone模式：使用Spark自带的集群资源管理器，效率低
  • Yarn模式：由Hadoop Yarn为Spark进行CPU和内存资源的调度
  • Mesos模式：使用Mesos作为集群资源管理器

Spark单机版和Hadoop伪分布式可以交互，访问HDFS文件；Spark集群部署模式也可以和Hadoop集群部署模式相互访问 

（二）在PySpark中运行代码

PySpark是一个交互式的执行环境（Spark Shell也是一个交互式的执行环境，但它是Scala语言）

开机启动进入Linux环境，Shell中输入命令进入PySpark环境：

pyspark --master <master-url>   # url不同，分别进入四种不同的环境
 
# 直接pyspark回车，为local[*]模式

Spark的运行模式取决于传递给 SparkContext 的 master-url 的值。Master URL可以是以下任一种形式：

• local：本地运行模式。用一个worker线程本地化运行spark（完全不并行，单线程）
• local[*]（默认）：也是单机环境，但不是单线程（本地化，但有一定的并行程度）。使用逻辑CPU个数（整个物理CPU个数*每个CPU核数）的线程来本地化运行spark
• local[K]：使用K个worker线程本地化运行spark（理想情况下，K应根据运行机器的CPU核数设定）
• spark://HOST:PORT：集群模式。HOST为主机名（如localhost），PORT是端口号（默认端口是7077）。连接到指定的Spark StandAlone master（自带集群资源管理器）
• mesos://HOST:PORT：连接到指定的mesos集群，默认端口是5050
• yarn模式：集群模式，资源调度管理器为yarn。取决于运行Spark时的driver节点（SparkContext）建在哪里：
  • yarn-client（yarn即默认为yarn-client）：driver节点在客户端机器上。用在程序开发人员调试程序时用，以客户端模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到
  • yarn-cluster：driver节点在集群中某台机器上。用在企业产品生产上线时用，以集群模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到

在Spark中采用本地模式启动PySpark的命令主要包含以下参数：

• --master：表示当前的pyspark要连接到哪个master。如果是local[*]，就是使用本地模式启动pyspark，其中括号内的星号表示需要使用几个CPU核心，也就是启动几个线程模拟Spark集群
• --jars：用于把相关的jar包添加到classpath中。如果有多个jar包，可以使用逗号分隔符连接它们

cd /usr/local/spark
./bin/pyspark --master local[4]   # 采用本地模式，在4个CPU核心上运行pyspark
./bin/pyspark --master local[4] --jars code.jar    # 可以在classpath中添加code.jar

执行 pyspark --help 命令，获取完整的选项列表：

cd /usr/local/spark
./bin/pyspark --help

执行如下命令启动pyspark（默认是local模式）：

./bin/pyspark

启动pyspark成功后在输出信息的末尾可以看到 >>> 的命令提示符。使用命令 exit() 退出pyspark 

（三）开发Spark独立应用程序 

1、WordCount程序 

# WordCount.py  统计文本文件中包含a的行的个数和b的行的个数
from pyspark import SparkConf, SparkContext
 
conf = SparkConf().setMaster("local").setAppName("My App")  # 生成配置的上下文信息
# MasterURL取值为local模式；通过网页查看管理时可以看到应用名称为My App
 
sc = SparkContext(conf = conf)   # 生成SparkContext对象
 
logFile = "file:///usr/local/spark/README.md"   # 若是本地文件，是file:///
 
logData = sc.textFile(logFile,2).cache()  # 把文本文件加载进来生成RDD
# RDD里包含很多元素，每个元素对应一行文本
 
numAs = logData.filter(lambda line: 'a' in line).count()   # 过滤出所有包含单词a的行
# lambda为匿名函数
# 把包含单词a的行全过滤出来，放在一个新的RDD中，再.count()统计
 
numBs = logData.filter(lambda line: 'b' in line).count()   # 过滤出所有包含单词b的行
 
print('Lines with a:%s, Lines with b:%s' % (numAs,numBs))

对于这段Python代码，可以直接使用如下命令执行：

cd /usr/local/spark/mycode/python
python3 WordCount.py

2、通过spark-submit运行程序

spark-submit
--master <master-url>
--deploy-mode <deploy-mode>   # 部署模式
...   # 其它参数
<application-file>   # Python代码文件
[application-arguments]   # 传递给主类的主方法的参数

执行 spark-submit --help 命令，获取完整的选项列表：

cd /usr/local/spark
./bin/spark-submit --help

上述代码以这种方式运行：

# 通过spark-submit提交到Spark中运行
/usr/local/spark/bin/spark-submit /usr/local/spark/mycode/python/WordCount.py
 
# 在命令中间使用“\”符号，把一行完整命令人为断开成多行进行输入
/usr/local/spark/bin/spark-submit \
/usr/local/spark/mycode/python/WordCount.py

为了避免其他多余信息对运行结果干扰，可以修改log4j的日志信息显示级别，从 log4j.rootCategory = INFO, console 改成 log4j.rootCategory = ERROR, console

（四）Spark集群环境搭建

假设有3台机器搭建集群：Master、Slave01、Slave02，且在搭建Spark集群之前，Hadoop集群的构建已经完成（Hadoop 2.7分布式集群环境搭建_厦大数据库实验室博客）

1、Spark+HDFS运行架构 

• Hadoop集群两大核心组件：NameNode（一个）、DataNode（多个），即一主多从
• Spark集群：Driver Node（一个），Worker Node（多个，负责具体任务计算，且实行数据的本地化处理，数据在哪Worker Node就在哪），即一主多从

一台机器上，既部署了Hadoop的DataNode，又部署了Spark的Worker Node，即HDFS里的DataNode和Spark的Worker Node共存。这样Spark的Worker Node可以对Hadoop的DataNode数据进行本地化计算

2、搭建Spark集群 

主节点为master，在master节点上安装Spark，和单机时步骤一样：

# 解压安装包spark-2.4.0-bin-without-hadoop.tgz至路径/usr/local
# usr是unix software resource
sudo tar -zxf ~/Downloads/spark-2.4.0-bin-without-hadoop.tgz -C /usr/local/
cd /usr/local
sudo mv ./spark-2.4.0-bin-without-hadoop/ ./spark  # 更改文件夹名
sudo chown -R hadoop ./spark   # 此处hadoop为系统用户名，把spark目录权限赋予hadoop用户

再执行以下命令：

# 在master节点主机的终端中执行
vim ~/.bashrc   # 隐藏文件
 
# 在.bashrc添加如下配置
export SPARK_HOME = /usr/local/spark
export PATH = $PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
 
# 运行source命令，使配置生效
source ~/.bashrc

接下来配置从节点slaves文件：

# 将slaves.template拷贝到slaves
cd /usr/local/spark
cp ./conf/slaves.template ./conf/slaves
 
# 编辑./conf/slaves，设置WorkerNode，把默认内容localhost替换成如下内容，一行一个
# 主机名称，从节点位于这两个主机上
slave01
slave02
 
# 配置spark-env.sh文件
cp ./conf/spark-env.sh.template ./conf/spark-env.sh   # 将spark-env.sh.template拷贝到spark-env.sh
 
# 编辑spark-env.sh，添加如下内容
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)   # 完成Spark和Hadoop的挂接
export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop   # 说明Hadoop相关配置信息的目录
export SPARK_MASTER_IP=192.168.1.104   # 设置Spark管家节点的IP地址
 
# 将master主机上的/usr/local/spark文件夹复制到各个节点上
cd /usr/local/
tar -zcf ~/spark.master.tar.gz ./spark  # 把主节点spark安装目录打包成文件
cd ~
scp ./spark.master.tar.gz slave01:/home/hadoop  # 把压缩包拷贝到两个从节点上
scp ./spark.master.tar.gz slave02:/home/hadoop
 
# 在从节点目录下执行解压缩操作
sudo rm -rf /usr/local/spark/
sudo tar -zxf ~/spark.master.tar.gz -C /usr/local
sudo chown -R hadoop /usr/local/spark   # 赋权限

3、开启和关闭Spark集群

开启：在master节点主机上运行

# 首先启动Hadoop集群
# Spark集群与Hadoop集群是搭配使用，Hadoop存数据，Spark计算
cd /usr/local/hadoop/
sbin/start-all.sh
 
# 启动master节点，在master节点主机上运行如下命令
cd /usr/local/spark/
sbin/start-master.sh
 
# 启动所有Slave节点，在master节点主机上运行（启动从节点是在主节点上启动）
sbin/start-slaves.sh
 
# 在master主机上打开浏览器，访问http://master:8080

关闭Spark集群：

# 关闭master节点（master节点上运行）
sbin/stop-master.sh
 
# 关闭Worker节点（master节点上运行）
sbin/stop-slaves.sh
 
# 关闭Hadoop集群
cd /usr/local/hadoop/
sbin/stop-all.sh

（五）在集群上运行Spark应用程序

1、启动Spark集群

以下命令均在master节点上运行： 

启动Hadoop集群：

cd /usr/local/hadoop/
sbin/start-all.sh

启动Spark的master节点和所有slaves节点：

cd /usr/local/spark/   # 进入spark安装目录
sbin/start-master.sh   # 启动主节点
sbin/start-slaves.sh   # 启动从节点

2、采用独立集群管理器（StandAlone）

（1）在集群中运行应用程序JAR包：需要把spark://host:port作为主节点参数传递给spark-submit

用一个程序提交给集群去算Π的值：

cd /usr/local/spark/
bin/spark-submit \
--master spark://master:7077 \   # 连接到Standalone独立集群模式，使用自带的集群资源管理器
/usr/local/spark/examples/src/main/python/pi.py 2>&1 | grep "Pi is roughly"  # 过滤出有用信息
# 结果为Pi is roughly 3.1415926

（2）在集群中运行Pyspark： 

cd /usr/local/spark/
bin/pyspark --master spark://master:7077  # Standalone模式连接到Spark集群
 
# 进入交互式运行环境
textFile = sc.textFile("hdfs://master:9000/README.md")   # 把底层文件（分布式文件系统hdfs里的文件）加载进来生成RDD
# hdfs是hdfs://，本地文件是file:///
textFile.count()   # 统计有多少行
textFile.first()   # 取出第一行内容

 运行后查看集群信息（用户在独立集群管理Web界面查看应用的运行情况）：http://master:8080/

3、采用Hadoop YARN管理器

• yarn-client：调试，在客户端建指挥所，客户端提交应用程序后不能关闭
• yarn-cluster：在集群里建指挥所，客户端提交应用程序后可以关闭客户端 

（1）spark-submit：确保Hadoop集群已经启动

cd /usr/local/spark/
bin/spark-submit \
--master yarn-client \   # yarn-client用来调试
/usr/local/spark/examples/src/main/python/pi.py

运行后，根据在shell中得到的输出结果地址查看（tracking URL）。复制结果地址到浏览器，点击查看Logs，再点击stdout，即可查看结果

（2）在集群中运行pyspark：用pyspark连接到采用Yarn作为集群管理器的集群上（交互式）

bin/pyspark --master yarn  # 默认yarn-client模式
 
# 假设HDFS根目录下已存在一个文件README.md，在pyspark环境中执行相关语句
textFile = sc.textFile("hdfs://master:9000/README.md")  
textFile.count()   # 统计RDD有多少元素
textFile.first()   # 取出第一行内容

在Hadoop Yarn集群管理Web界面（http://master:8088/cluster）查看所有应用的运行情况