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：是否仅根据文件名而不是完整路径来检查新文件，默认为FalseKafka源
• 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.评估模型拟合程度的准则和方法