一、Spark介绍

hadoop生态圈：

1.1 Scala和PySpark

（1）Scala 是一门多范式（multi-paradigm）的编程语言，设计初衷是要集成面向对象编程和函数式编程的各种特性。

Scala 运行在 Java 虚拟机上，并兼容现有的 Java 程序。

Scala 源代码被编译成 Java 字节码，所以它可以运行于 JVM 之上，并可以调用现有的 Java 类库。

（2）Apache Spark是用 Scala编程语言 编写的。为了用Spark支持Python，Apache Spark社区发布了一个工具PySpark。使用PySpark，也可以使用Python编程语言中的 RDD 。

（3）PySpark提供了 PySpark Shell，它将Python API链接到spark核心并初始化Spark上下文。将Python与Spark集成就对数据科学研究更加方便。

Spark的开发语言是Scala，这是Scala在并行和并发计算方面优势的体现，这是微观层面函数式编程思想的一次胜利。此外，Spark在很多宏观设计层面都借鉴了函数式编程思想，如接口、惰性求值和容错等。

1.2 Spark原理

Spark是业界主流的大数据处理利器。

分布式：指的是计算节点之间不共享内存，需要通过网络通信的方式交换数据。

Spark 是一个分布式计算平台。Spark 最典型的应用方式就是建立在大量廉价的计算节点上，这些节点可以是廉价主机，也可以是虚拟的 Docker Container（Docker 容器）。

Spark 的架构图中：

Spark 程序由 Manager Node（管理节点）进行调度组织

由 Worker Node（工作节点）进行具体的计算任务执行

最终将结果返回给 Drive Program（驱动程序）。

在物理的 Worker Node 上，数据还会分为不同的 partition（数据分片），可以说 partition 是 Spark 的基础数据单元。

图1 Spark架构图

Spark 计算集群能够比传统的单机高性能服务器具备更强大的计算能力，就是由这些成百上千，甚至达到万以上规模的工作节点并行工作带来的。

1.3 一个具体栗子

那在执行一个具体任务的时候，Spark 是怎么协同这么多的工作节点，通过并行计算得出最终的结果呢？这里我们用一个任务来解释一下 Spark 的工作过程。

一个具体任务过程：

（1）先从本地硬盘读取文件 textFile；

（2）再从分布式文件系统 HDFS 读取文件 hadoopFile；

（3）然后分别对它们进行处理；

（4）再把两个文件按照 ID 都 join 起来得到最终的结果。

在 Spark 平台上处理这个任务的时候，会将这个任务拆解成一个子任务 DAG（Directed Acyclic Graph，有向无环图），再根据 DAG 决定程序各步骤执行的方法。从图 2 中可以看到，这个 Spark 程序分别从 textFile 和 hadoopFile 读取文件，再经过一系列 map、filter 等操作后进行 join，最终得到了处理结果。

图2 某Spark程序的任务有向无环图

最关键的过程是要理解哪些是可以纯并行处理的部分，哪些是必须 shuffle（混洗）和 reduce 的部分：这里的 shuffle 指的是所有 partition 的数据必须进行洗牌后才能得到下一步的数据，最典型的操作就是图 2 中的 groupByKey 操作和 join 操作。以 join 操作为例，必须对 textFile 数据和 hadoopFile 数据做全量的匹配才可以得到 join 后的 dataframe（Spark 保存数据的结构）。而 groupByKey 操作则需要对数据中所有相同的 key 进行合并，也需要全局的 shuffle 才能完成。

与之相比，map、filter 等操作仅需要逐条地进行数据处理和转换，不需要进行数据间的操作，因此各 partition 之间可以完全并行处理。

在得到最终的计算结果之前，程序需要进行 reduce 的操作，从各 partition 上汇总统计结果，随着 partition 的数量逐渐减小，reduce 操作的并行程度逐渐降低，直到将最终的计算结果汇总到 master 节点（主节点）上。可以说，shuffle 和 reduce 操作的触发决定了纯并行处理阶段的边界。

图3 被shuffle操作分割的DAG stages

注意：

（1）shuffle 操作需要在不同计算节点之间进行数据交换，非常消耗计算、通信及存储资源，因此 shuffle 操作是 spark 程序应该尽量避免的。shuffle可以理解为一个串行操作，需要等到在此之前的并行工作完成之后才可以顺序开始。

（2）简述Spark 的计算过程：Stage 内部数据高效并行计算，Stage 边界处进行消耗资源的 shuffle 操作或者最终的 reduce 操作。

二、安装方式

Windows 10：不适合开发程序，因为不支持命令行工具、隐藏坑较多、解决方案的资料较少

Windows Subsystem Linux (WSL)：需要安装较多软件和配置较多环境变量，非常麻烦

ubuntu / CentOS：未尝试，但与WSL比较相似

docker：简单、高效、可迁移

docker方式（在ubuntu环境）：

安装docker：curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun；

拉取镜像：docker pull jupyter/pyspark-notebook

创建容器：

docker run \
    -d \
    -p 8022:22 \
    -p 4040:4040 \
    -v /home/fyb:/data \
    -e GRANT_SUDO=yes \
    --name myspark \
    jupyter/pyspark-notebook

配置docker容器的SSH登录

安装openssh-server等常用软件：apt update && apt install openssh-server htop tmux

设置允许root通过ssh登录：echo "PermitRootLogin yes" >> /etc/ssh/sshd_config

重启ssh服务：service ssh --full-restart，设置root用户密码：passwd root

测试docker容器内的ssh是否设置成功：ssh root@127.0.0.1 -p 8022

容器内的配置python环境：

以root用户登录SSH会话后，安装python依赖工具：apt install pip

安装PySpark依赖包：pip3 install pyspark numpy pandas tqdm

测试是否正确安装并执行了全部修改：python3 /usr/local/spark/examples/src/main/python/pi.py

三、测试是否安装成功

四、Spark程序的模块分类

五、数据处理任务

5.1 使用Python链接Spark环境

import pandas as pd
from pyspark.sql import SparkSession
# 创建spark应用 mypyspark
spark = SparkSession.builder.appName('mypyspark').getOrCreate()

5.2 创建dateframe数据

这里和pandas等工具类似，创建表时注意这里的表头组成的list列表，放在数据的后面。

test = spark.createDataFrame([('001','1',100,87,67,83,98), ('002','2',87,81,90,83,83), ('003','3',86,91,83,89,63),
                            ('004','2',65,87,94,73,88), ('005','1',76,62,89,81,98), ('006','3',84,82,85,73,99),
                            ('007','3',56,76,63,72,87), ('008','1',55,62,46,78,71), ('009','2',63,72,87,98,64)],                           
                             ['number','class','language','math','english','physic','chemical'])
test.show(5)

5.3 用spark执行以下逻辑：找到数据行数、列数

# 查看表前2行
test.head(2)
test.describe().show()
# 列出表头属性
test.columns
# 列出第一行的数据
test.first()
# 数据大小  shape
print('test.shape: %s行 %s列'%(test.count(), len(test.columns)))
# 上面打印出 test.shape: 9行 7列

5.4 用spark筛选class为1的样本

这里可以使用df.filter或df.where：

# 方法一
test.filter(test['class'] ==1).show()
# 方法二
test.filter('class == 1' ).show()

5.5 用spark筛选language >90 或 math> 90的样本

test.filter('language>90 or math>90').show()
test.where('language>90 or math>90').show()
test.filter((test['language']>90)|(test['math']>90)).show()

任务汇总：