Spark编程基础（Python版）- 林子雨期末复习攻略

第一章 大数据技术概述

大数据

大数据特征：

大量化Volume

快速化Velocity

多样化Variety

价值密度低Value

基本处理流程

1. 数据采集
2. 存储管理
3. 处理分析
4. 结果呈现

大数据关键技术

1. 数据采集
2. 数据存储与管理
3. 数据处理与分析
4. 数据隐私与安全
5. 数据可视化

计算模式

批处理计算

流计算

图计算

查询分析计算
| 大数据计算模式 |解决问题 |代表产品 |
| 批处理计算 | 组队打规模数据的批量处理| MapReduce/Spark等 |
| 流计算 | 针对流数据的实时计算 | Storm/S4/Streams/Puma等 |
| 图计算 | 针对大规模图结构数据的处理 | Pregel/GraphX/Giraph |
| 查询分析计算 | 大规模数据的存储管理和查询分析 | Dremel/Hive/Gassandra/Impala等 |

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与关系型数据库数据之间的导入导出

名称节点：NameNode
数据节点：DataNode

Hadoop与Spark

spark的设计遵循：一个软件栈满足不同应用场景的理念

1. 即席查询 SparkSQL
2. 流计算 SparkStreaming
3. 机器学习MLlib
4. 图计算GraphX

对比优点：

1. Spark计算模式也属于MapReduce，但不局限于Map和Reduce操作，还提供了多种数据集操作类型，编程模型比MapReduce更灵活
2. Spark提供了内存计算，中间结果直接存放到内存中，带来了更高的迭代运算效率
3. Spark基于DAG的任务调度执行机制，要优于MapReduce的迭代执行机制

对比缺点：

1. 表达能力有限
2. 磁盘I/O开销大
3. 延迟高

spark与hadoop的统一部署：可以在YARN上统一部署各个计算框架。

第二章 Spark的设计与运行原理

Spark：发展、特点

Spark最初由美国加州大学伯克利分校的AMP实验室于2009年开发，是基于内存计算的大数据并行计算框架，可用于构建大型的、低延迟的数据分析应用程序。
特点：

1. 运行速度快
2. 容易使用
3. 通用性
4. 运行模式多样

Spark生态系统

大数据处理3种类型：

1. 复杂的批量数据处理：时间跨度通常在数十分钟到数小时之间、MapReduce、Hive、生态组件SparkCore
2. 基于历史数据的交互式查询：时间跨度通常在数十秒到数分钟之间、生态SparkSQL
3. 基于实时数据流的数据处理：时间跨度通常在数百毫秒到数秒之间、生态SparkStreaming、StrucuredStreaming

spark的生态系统：

1. 访问和接口：SparkStreaming、BlinkDB、SparkSQL、GraphX、MLbase、MLlib
2. 处理引擎：SparkCore
3. 存储：Tachyon、HDFS、S3
4. 资源管理调度：Mesos、HadoopYARN

Spark运行框架

基本概念：

1. RDD：是弹性分布式数据集，是分布式内存的一个抽象概念，提供了一种高度受限的共享内存模型
2. DAG：有向无环图，反映RDD之间的依赖关系
3. Executor：是运行在工作节点上的一个进程，负责运行任务，并为应用程序存储数据
4. 应用（Application）：用户编写的Spark应用程序
5. 任务（Task）：运行在Executor上的工作单元
6. 作业（Job）：一个作业包含多个RDD及作用于相应RDD上的各种操作
7. 阶段（Stage）：是作业的基本调度单位，一个作业会分为多组任务，每组任务被称为“阶段”，或者也被称为“任务集“

架构设计：

集群资源管理器（可以是Spark自带的资源管理器，也可以是YARN或Mesos等资源管理器框架）

运行作业任务的工作节点

每个应用的任务控制节点

每个工作节点

Spark采用“主从架构”，包含一个Master和若干个Worker

Spark采用Executor的优点：

利用多线程来执行具体任务，减少任务的启动开销

Executor中有一个BlockManager存储模块，会将内存和磁盘共同作为存储设备

Spark运行基本流程：Spark任务调度流程

spark的任务调度流程分为RDDObject、Stage、Task、Worker四个部分。

1. RDDObject是Spark中的基础数据结构，它包含了数据的逻辑结构和物理结构。
2. Stage是Spark中的任务级别，它包含了一组Task，每个Task是独立的计算单元。
3. Task是Spark中的任务单元，它包含了一组操作，每个操作是一个独立的计算单元。
4. Worker是Spark中的工作节点，它负责执行Task。

RDD的设计与运行原理：转换操作、行动操作、DAG、宽依赖、窄依赖

转换操作：（map、filter、groupBy、join等）接受RDD并返回RDD

行动操作：（count、collect等）接受RDD但是返回非RDD

DAG：构建起fileRDD和filterRDD之间的依赖关系，形成DAG图

宽依赖：groupByKey、sortByKey

窄依赖：map、filter、union

Spark部署方式

四种不通类型的部署方式：

1. 单机：Local
2. 集群：Standalone、Spark on Mesos、Spark on YARN

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

第三章Spark环境和使用方法

Spark的基本环境配置

1. Linux
2. Hadoop
3. JDK
4. Spark

环境变量

1. JAVA_HOME
2. HADOOP_HOME
3. PYTHONE_HOME
4. SPARK_HOME
5. PYSPARKPATH
6. PYSPARK_PYTHON
7. PATH

Yarn-client模式和Yarn-cluster模式

Yarn-client：Spark采用YARN集群模式，以客户端模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到，当用户提交了作业后，不能关掉client，Driver Program驻留在Client中，负责调度作业的执行；该模式适合运行交互类型的作业，常用于开发测试阶段。

Yarn-cluster：Spark采用YARN集群模式，以集群模式连接YARN集群，集群的位置可以在HADOOP_CONF_DIR环境变量中找到，当用户提交了作业后，不能关掉client，Driver Program驻留在Client中，负责调度作业的执行；该模式不适合运行交互类型的作业，常用于企业生成环境。

Spark与Hadoop的相互关系

Hadoop（伪分布式）+Spark（Local），Hadoop和Spark可以相互协作，由Hadoop的HDFS、HBase等组件负责数据的存储和管理，由Spark进行计算。

第四章RDD编程

1、RDD的创建：文件、并行集合

1. 本地加载创建
   lines = sc.textFile(“file:///usr/local/spark/1.txt”)
   lines.foreach(print)
2. 分布式文件系统加载创建下面三选一即可
   lines = sc.textFile(“hdfs:///usr/local/spark/1.txt”)
   lines = sc.textFile(“/usr/local/spark/1.txt”)
   lines = sc.textFile(“/1.txt”)
   lines.foreach(print)
3. 通过并行集合创建
   array = [1,2,3,4,5]
   rdd = sc.parallelize(array)
   rdd.foreach(print)

2、RDD的执行过程

①准备资源
②创建Driver和Executor节点
②然后将应用程序的数据处理逻辑分解成一个一个的计算任务task。
③然后将任务task发到【已经分配资源】的计算节点executor上, 按照指定的计算模型进行数据计算。最后得到计算结果

3、RDD操作

filter:筛选出满足条件的元素

map：将每个元素传递到函数func中

flatMap：与map相似，但每个输入元素都可以映射到0或多个输出结果

groupBy：应用于（K,V）键值对的数据时，返回一个新的数据集

reduceByKey：应用于（K,V）键值对的数据时，返回一个新的数据集，其中每个值是将每个key传递到函数中进行聚合后的结果

count：统计元素个数

collect：以数组返回所有元素

first：返回第一个元素

take：以数组的形式返回数据集中的前n个元素

reduce：通过函数func聚合数据集中的元素

foreach：将数据集中的每个元素传递到函数func中

4、分区的作用

增加并行度、减少通信开

5、键值对RDD:创建、键值对转换操作reduceBykey、groupByKey、sortByKey、join等

创建：lines.flatMap(lambda x:x.split(“,”)),map(lambda x:(x,1))

reduceBykey: 合并相同的键的值

groupByKey：对相同键的值进行分组

sortByKey：根据key排序

join：表示内连接，对于给定的两个数据集进行合并，看哪个对哪个进行连接，如（B对A连接）就是在A的基础上[0][1]进行B的数据集

第五章SparkSQL

1、Shark

shark提供了类似于关系型SQL的查询语言—–HiveQL，用户可以通过HiveQL语句快速实现简单的MapReduce统计，Hive自身可以自动将HiveQL语句快速转换成MapReduce任务进行运行。

2、SparkSQL

用户需要从不通数据源执行各种操作，包括结构化喝非结构化数据；其次用户需要执行高级分析，如机器学习和图处理，在实际大数据中，经常需要融合关系查询和复杂分析算法，但是，一直以来都缺少这样的系统。

3、DataFrame：创建、保存、常用操作

创建SparkSession对象

from pyspark import SparkContext，SparkConf
from pyspark.sql import SparkSession
spark = SparkSession.builder.config(conf = SparkConf()).getOrCreate()

创建DataFrame：

spark.read.text(“xxxxx.txt”)

spark.read.json(“xxxxx.json”)

spark.read.parquet(“xxxxx.parquet”)

spark.read.format(“text”).local(“xxxxx.txt”)

spark.read.format(“json”).local(“xxxxx.json”)

spark.read.format(“parquet”).local(“xxxxx.parquet”)
df = spark.read.xxxx(“xxxxx.xxx”)

保存：

df.write.text(“xxxxx.txt”)

df.write.json(“xxxxx.json”)

df.write.parquet(“xxxxx.parquet”)

df.write.format(“text”).save(“xxxxx.txt”)

df.write.format(“json”).save(“xxxxx.json”)

df.write.format(“parquet”).save(“xxxxx.parquet”)

常用操作：

printSchema():打印DataFrame的模式

select：从DataFrame的选取部分列的数据

filter：实现条件查询

groupBy：对记录进行分组

sort：对记录进行排序（desc降序，asc升序）

show：打印数据

4、从RDD转换得到DataFrame

利用反射机制推断RDD模式：利用反射机制来推断包含特定类型对象的RDD的模式，适用于数据结构已知的RDD转换
使用编程方式定义RDD模式：使用编程接口构造一个模式，并将其应用在已知的RDD上，适用于数据结构未知的RDD转换

5、SparkSQL编程 113页的实验

第六章Spark Streaming

1、批量计算与实时计算

流计算需求：

高性能

海量式

实时性

分布式

易用性

可靠性
流计算框架：

IBM InfoSphere

IBM StreamBase

Twitter Storm

Yahoo！S4

Facebook Puma

DStream

银河流数据处理平台

Super Mario
流计算处理流程：
1.数据实时采集
2.数据实时计算
3.实时查询服务

2、流计算框架Storm和Spark Streaming的对比

Spark Streaming和Storm最大的区别在于，Spark Streaming无法实现毫秒级的流计算，而storm可以实现毫秒级的流计算
Spark Streaming难以满足实时性要求非常高的场景，但足以胜任其他实时计算场景，相比之下，Storm处理数据单元为元组，只会产生极小的延迟。
Spark Streaming构建在Spark Core上，因为spark低延迟执行引擎可以用于实时计算，相比于Strom，RDD数据集更易于进行高效的容错处理。
Spark Streaming采用小批量处理的方式，使得它可以同事兼容批量和实时数据处理的逻辑和算法，便于在一些需要历史数据和实时数据联合分析的特定应用场景使用。

3、编写Spark Streaming程序的基本步骤

1.通过创建输入DStream来定义输入源
2.通过对DStream应用转换操作和输出操作来定义流计算
3.通过StreamingContext对象的start()方法来开始接收数据和处理流程
4.通过调用StreamContext对象的awaitTermination()方法来等待流计算经常结束，或者也可以通过调用StreamingContext对象的stop()对象来手动结束流计算流程

4、基本数据源：文件流、套接字流（Scoket）和RDD队列流（编程方法）

文件流：
在pyspark中创建：
from pyspark import SparkContext
from pyspark.streaming imoprt StreamingContext
ssc = StreamingContext(sc,10)
lines = ssc.textFile(‘file:///usr/local/logfile’)
words = lines.flatMap(lambda x:x.split(’ ‘))
wordCounts = words.map(lambda x:(x,1)).reduceByKey(lambda a,b:a+b)
wordCounts.pprint()
ssc.start()
ssc.awaitTermination()
独立应用程序方式创建文件流
from pyspark import SparkContext,SparkConf
from pyspark.streaming imoprt StreamingContext
sc = SparkContext(conf = SparkConf.setAppName(“test”).setMaster(‘local[2]’))
ssc = StreamingContext(sc,10)
lines = ssc.textFile(‘file:///usr/local/logfile’)
words = lines.flatMap(lambda x:x.split(’ '))
wordCounts = words.map(lambda x:(x,1)).reduceByKey(lambda a,b:a+b)
wordCounts.pprint()
ssc.start()
ssc.awaitTermination()
套接字流：
ssc.socketTextStream()
RDD队列流:
queueStream()

5、高级数据源：Kafka及相关概念、Flume

Kafka及相关概念

Broker:Kafka集群包含一个或多个服务器，这些服务器被称为Broker

Topic：每条发送到Kafka集群的消息都有一个类型，这个类别被称为Topic

partition：是物理上的概念，每个Topic包含一个或多个Partition

producer：负责发布消息到Kafka Broker

consumer：消息消费者，向KafkaBorker读取消息的客户端

consumerGroup:每个consumer属于一个特定的consumerGroup，可为每个consumer指定groupname，若不指定groupname，则属于默认的group

6、Dstream转换操作

Dstream转换包含无状态转换操作和有状态转换操作

有状态：滑动窗口转换操作和updateStateByKey操作
前者以一个时间阶段为滑动窗口进行操作，后者则用来跟踪每个键的状态变化。

无状态：不会记录历史状态信息，每次对新的批次数据进行处理时，只会记录当前批次数据的状态

第七章Structured Streaming

1、Structured Streaming基本概念

Structured Streaming的关键思想是将实时数据流视为一张正在不断添加数据的表，这种新的流处理模型与批处理模型十分类似。

2、Structured Streaming处理模式

1.微批处理模型
Structured Streaming默认使用微批处理模型，这意味着Spark流计算引擎会定期检查流数据源，并对自上一批次结束后到达的新数据执行批量查询。
2.持续处理模型
微批处理的数据延迟对于大多数实际的流式工作负载已经足够了，但一些场景确实需要更低的延迟

3、Structured Streaming编程基本步骤

（1）导入pyspark模块
（2）创建SparkSession对象
（3）创建输入数据源
（4）定义流计算过程
（5）启动流计算并输出结果

4、SparkSQL、Spark Streaming、Structured Steaming的关系

Structured sreaming处理的数据与Spark Streaming一样，也是源源不断的数据流，它们之间的区别在于，Spark Streaming采用的数据抽象是DStream(本质上就是一系列RDD),而Structured Streaming采用的数据抽象是DataFrame。Structured Streaming可以使用Spark SQL的DataFrame/Dataset来处理数据流。虽然Spark SQL也是采用DataFrame作为数据抽象，但是，SparkSOL只能处理静态的数据，而Structured Steaming可以处理结构化的数据流。这样一来,Struetured Streaming就将Spark SQL和SparkStreaming二者的特性结合起来。Structured Streaming可以对DataFrame/Dataset应用前面章节提到的名种操作，包括select、where、groupBy、map、filter、fatMap等。此外，Spark Streaming只能实现秒级的实时响应，而Stuctured Streaming由于采用了全新的设计方式，采用微批处理模型时可以实现 100毫秒级别的实时响应，采用持续处理模型时可以支持毫秒级的实时响应。

5、输入源：File源、Kafka源、Socket源、Rate源

File源

path:输入路径的目录，所有文件格式通用

maxFilesTrigger：每个触发器中要处理的最大新文件数（默认无最大值）

latestFirst:是否优先处理最新的文件，当有大量文件积压时，设置为True可以优先处理新文件，默认为False

fileNameOnly：是否仅根据文件名而不是完整路径来检查新文件，默认为False
Kafka源

assign：指定所消费的Kafka主题和分区

subscribe：订阅的Kafka主题，为逗号分隔的主题列表

subscribePattern：订阅的Kafka主题正则表达式，可匹配多个主题

kafka.bootstrap.servers:Kafka服务器的列表，逗号分割的“host:post”列表

startingOffsets：起始位置偏移量

endingOffsets：结束位置偏移量

failOnDataLoss：布尔值，表示是否在Kafka数据可能丢失时触发流计算失败。
Socket源

host：主机ip地址或域名，必须设置

port：端口号，必须设置

includeTimestamp：是否在数据行内包含时间戳
Rate源

rowsPerSecond：每秒产生多少行数据，默认为1

rampUpTime：生成速度达到rowsPerSecond需要多少启动时间，使用比秒更精细的粒度将会截断为整数秒，默认为0秒

numPartitions：使用的分区数，默认为Spark的默认分区数

6、输出模式

Append模式：只有结果表中自上次触发间隔后增加的新行，才会被写入外部存储器

Complete模式：已更新的完整的结果表可被写入外部存储器

Update模式：只有自上次触发间隔后结果表中发生更新的行，才会被写入外部存储器

第八章Spark MLib

1、MLib

算法工具：常用的学习算法，如分类、回归、聚类和协同过滤；
特征化工具：特征提取、转化、降维和选择工具；
流水线(Pipeline)：用于构建、评估和调整机器学习工作流的工具;
持久性：保存和加载算法、模型和管道;
实用工具：线性代数、统计、数据处理等工具。

2、基本数据类型

本地向量分为稠密向量（DenseVector）和稀疏向量（SparseVector）

标注点是一种带有标签的本地向量，可以是稠密向量也可以是稀疏向量

本地矩阵具有整型的行、列索引值和双精度浮点型的元素值，并存储在单机上

3、机器学习流水线

1.DataFrame
2.转换器（Transformer）
3.评估器（Estimator）
4.流水线（Pipeline）
5.参数（Parameter）

4、特征提取、转换和选择

1.特征提取：从原始数据中抽取特征
2.特征转换：缩放、转换和修改特征
3.特征选择：从较大特征集中选取特征子集

5、分类、聚类

分类
逻辑斯蒂回归分类器
P（Y = 1|x）= exp(w * x+b)/1+exp(wx+b)
P（Y = 0|x）= 1/1+exp(wx+b)
决策树分类器
1.信息增益：
G（D,A）=H(D)-H(D|A)
2.信息增益比
Gr（D,A）=G(D,A)/HA(d)
3.基尼指数：
Gini§=k=1,pk(1-pk)=1-k=1,pk2
聚类
1.每个蔟至少包含一个数据对象
2.每个数据对象仅属于一个蔟
K-Means聚类算法
1.根据给定的k值，选取k个样本点作为初始划分中心。
2.计算所有样本店到每一个划分中心的距离，并将所有样本店划分到距离最近的划分中心
3.计算每个划分样本店的平均值，并将其作为新的中心
4.循环进行步骤（2）和（3）至最大迭代次数，或划分中心的变化小于某一预定义阈值
高斯混合模型（GMM）
P(x)=i=1,k++,wi*p(x|μi，累加i)

6、模型选择

1.待调优的算法或流水线
2.一系列参数表，是可选参数
3.评估模型拟合程度的准则和方法

作者：Blitzcrank007!