Spark是什么

整个Hadoop生态圈分为分布式文件系统HDFS、计算框架MapReduce以及资源调度框架Yarn。但是随着时代的发展，MapReduce其高强度的磁盘IO、网络通信频率以及写死了的使得其严重拖慢了整个生态圈的运行速度。Spark因此而生。

首先我们来讲一下原有的MapReduce是如拖慢节奏的：
对于一个统计词频的任务来说，map任务首先会分散统计在当前节点的词频信息，然后将这个临时信息落地到磁盘之中，之后reduce任务会从磁盘中读取这些数据进行混洗。混洗会把相同的键映射到同一个执行reduce任务的节点之中这个过程就会涉及到大量的网络通信。而Reduce任务执行完之后，会再次把数据落地到磁盘中。

可以看到一次MapReduce阶段就会经过至少两次网络IO+N次网络通信，这也是两个主要耗时的任务。特别是对于神经网络或者图计算这种需要频繁迭代的计算来说，这会涉及到大量的磁盘IO和网络通信。

Spark对此进行了改进，其通过DAG图和内存计算来改进上面的这些缺点。

DAG有向无环图

提到DAG有向无环图就需要先提及其基本组成元素RDD，又叫弹性分布式数据集。其会把所有中间环节的数据文件以某种统一的方式归纳、抽象出来，那么所有map与reduce就可以更流畅地衔接在一起。在具体的Spark开发过程中来讲，其就是对磁盘数据的描述信息，当遇到action操作的时候，Spark就会根据RDD中的信息对数据进行计算。

在RDD中主要有转换和行动两种操作。转换操作就会在多个RDD之间开始构建DAG有向无环图，其描述了RDD之间的关联关系。假如说，我们会从文本文件README.md中读取了数据创建了readme_rdd，这个就是有向无环图的首节点。之后我们加入要查找这个文件中包含“Python”这个单词的行，我们就会转换readme_rdd成python_rdd。

但是我们要记住一点：rdd是惰性求值的，转换只是记录对数据的操作关系的，并不会真正的进行运算。至于为什么这么做则是spark可以将一些操作合并到一起以减少计算数据的步骤。假如说我们启动rdd的行动操作，那么spark才会真正的根据这个DAG中的内容进行执行。

注：
但是MapReduce并没有完全被Spark取代，因为内存容量问题Spark对于某些问题可能不能正常处理

spark环境搭建

使用以下命令下载并搭建spark环境

wget --no-check-certificate https://dlcdn.apache.org/spark/spark-3.2.0/spark-3.2.0-bin-hadoop3.2.tgz
tar zxvf spark-3.2.0-bin-hadoop3.2.tgz

之后就可以直接使用了

注：
spar k需要运行在 jvm环境，所以需要 jdk1.8版本以上

Spark开发

pyspark使用

我们可以直接调用spark提供的pyspark来执行语句，其会默认初始化一个SparkContext对象sc，我们可以直接调用其方法创建rdd。

./bin/pyspark
lines = sc.textFile("README.md")
lines.count()

而我们想要写python脚本操作spark就需要使用spark提供的spark-submit组件：

./bin/spark-submit test.py

python

Spark初始化

要在python中使用Spark就需要先配置好SparkConf，其主要注意的是两个参数一个是Spark集群的Url以及这个驱动服务的名称。

from pyspark import SparkConf, rdd
from pyspark.context import SparkContext# spark配置
conf = SparkConf().setMaster("local").setAppName("app")

之后就可以根据这个SparkConf创建一个SparkContext对象。有了这个rdd我们就可以创建RDD并进行一系列操作了。

sc = SparkContext(conf=conf)

创建RDD

RDD的输入源主要有三种：

• 本地文件系统

file = sc.textFile("file:[绝对路径]")

或者我们也可以向下面的例子一样，直接使用相对路径创建RDD：

lines = sc.textFile("README.md")

• Amazon S3
  要在Spark中访问S3数据，首先就得把S3访问凭据设置为AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY环境变量

file = sc.textFile("s3n:[绝对路径]")

• HDFS

file = sc.textFile("hdfs://[ip:port]/[绝对路径]")

当然除了从磁盘读取，其也可以转换内存中的数据成为RDD：

memory_rdd = sc.parallelize(["pandas","china","Chinese","chinese food"])

数据的读取和保存

创建RDD必然离不开数据的读取，其主要有以下几种文件格式可以读取：

• 文本文件
• JSON
• SequenceFile
• protobuf

目前暂且只介绍前面两种常用的文件格式。

文本文件

对于本地的文本文件，我们直接使用刚刚的textFile就可以了。同时pyspark还支持我们对整个文件夹进行操作，其会读取整个文件夹，之后以文件名作为Key，文件内容作为Value进行创建输入RDD。下面是个例子：

读取目录,这样会形成键值对<文件名,文件内容>
dir_rdd = sc.wholeTextFiles("file:///home/ik/software/spark/python_demo")
print(dir_rdd.keys().collect()) # 输出键

输出如下：

['file:/home/ik/software/spark/python_demo/pydemo.py', 
'file:/home/ik/software/spark/python_demo/README.md', 
'file:/home/ik/software/spark/python_demo/pair.py', 
'file:/home/ik/software/spark/python_demo/LocalFile.py', 
'file:/home/ik/software/spark/python_demo/JsonTest.json', 
'file:/home/ik/software/spark/python_demo/JsonTest1.json', 
'file:/home/ik/software/spark/python_demo/SparkPlus.py']

跟我们下面的结果是一致的，只是其不会递归读入文件夹里面的内容。

而保存文件则是调用saveAsTextFile这个方法，其会在路径下创建一个目录。

dir_rdd.keys().saveAsTextFile("file:///home/ik/software/spark/python_demo")

这个目录下会有四个文件，之后红箭头所指的文件才是文件中的内容。

Json文件

Json文件的接口本质上可文本文件是一致的，只不过我们需要调用Python的加载器。

json_rdd = sc.textFile("file:///home/ik/software/spark/python_demo/JsonTest1.json")
data = json_rdd.map(lambda x:json.loads(x))
print(data.collect())

RDD的转换操作

RDD是只读的，所以其转换操作其实是从父RDD中抽取一部分数据经过转换形成新的RDD。这其中主要是涉及到两个方法map和filter。

• map:接收一个函数，把这个函数用于RDD的每个元素，将函数的返回结果作为RDD中元素的值

nums = sc.parallelize([1,2,3,4,5])
for number in nums.map(lambda x:x*x).collect():print(number)

• filter：接收一个函数，用于筛选RDD中的数据

python_line = lines.filter(lambda line:"Python" in line)
print(python_line.first())

• flatmap：其也接收一个函数，不过其会把原来的一个元素根据函数规则划分为两个元素

sentence = sc.parallelize(["hello world","hi world"])
words = sentence.flatMap(lambda line:line.split(" "))
print(words.collect())	# ['hello', 'world', 'hi', 'world']

RDD的行动操作

行动操作会真正的触发Spark的运算，主要是以下方法：

• count：返回计数结果

lines = sc.textFile("README.md")
print(lines.count())

• take：其接收一个整数操作，这个整数表示从RDD中抽取几个元素。之后我们可以对这些元素进行打印

# take获取10个数据
rdd_set = python_line.take(10)
for line in rdd_set:print(line)

• collect：其会获取RDD中的所有元素，但是要注意如果对象过大会导致内存爆掉

# 传递函数，主要是filter
def filter_func(lines):return "Python" in linespython_line2 = lines.filter(filter_func)
for line in python_line2.collect():print(line)

集合操作

集合操作主要如下所示，就不多介绍了：

rdd1 = sc.parallelize([1,3,5,7,9,1])
rdd2 = sc.parallelize([2,4,6,8,1])
# distinct 去重
print(rdd1.distinct().collect())
# intersection 运行时只返回重复的元素
print (rdd1.intersection(rdd2).collect())
# subtract 返回只存在第一个rdd而不存在于第二个rdd中的元素
print(rdd1.subtract(rdd2).collect())
# cartesian 计算笛卡尔积
print(rdd1.cartesian(rdd2).collect())

mysql读取

  # mysql 配置(需要修改)prop = {'user': 'xxx', 'password': 'xxx', 'driver': 'com.mysql.cj.jdbc.Driver'}# database 地址(需要修改)url = 'jdbc:mysql://host:port/database'# 读取表data = spark.read.jdbc(url=url, table='tb_newCity', properties=prop)# 打印data数据类型print(type(data))# 展示数据data.show()# 关闭spark会话spark.stop()