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

第一章 大数据技术概述

大数据

大数据特征：

• 大量化Volume
  
• 快速化Velocity
  
• 多样化Variety
  
• 代价密度低Value
  

根本处理流程

• 数据收罗
  
• 存储管理
  
• 处理分析
  
• 结果出现
  

大数据关键技术

• 数据收罗
  
• 数据存储与管理
  
• 数据处理与分析
  
• 数据隐私与安全
  
• 数据可视化
  

盘算模式

• 批处理盘算
  
• 流盘算
  
• 图盘算
  
• 查询分析盘算
  | 大数据盘算模式 |解决题目 |代表产品 |
  | 批处理盘算 | 组队打规模数据的批量处理| 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的设计遵照：一个软件栈满足不同应用场景的理念

• 即席查询 SparkSQL
  
• 流盘算 SparkStreaming
  
• 机器学习MLlib
  
• 图盘算GraphX
  

对比优点：

• Spark盘算模式也属于MapReduce，但不范围于Map和Reduce操作，还提供了多种数据集操作范例，编程模型比MapReduce更灵活
  
• Spark提供了内存盘算，中心结果直接存放到内存中，带来了更高的迭代运算效率
  
• Spark基于DAG的使命调度执行机制，要优于MapReduce的迭代执行机制
  

对比缺点：

• 表达能力有限
  
• 磁盘I/O开销大
  
• 耽误高
  

spark与hadoop的同一摆设：可以在YARN上同一摆设各个盘算框架。
第二章 Spark的设计与运行原理

Spark：发展、特点

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

• 运行速率快
  
• 轻易利用
  
• 通用性
  
• 运行模式多样
  

Spark生态体系

大数据处理3种范例：

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

spark的生态体系：

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

Spark运行框架

根本概念：

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

架构设计：

• 集群资源管理器（可以是Spark自带的资源管理器，也可以是YARN或Mesos等资源管理器框架）
  
• 运行作业使命的工作节点
  
• 每个应用的使命控制节点
  
• 每个工作节点
  
• Spark采用“主从架构”，包罗一个Master和若干个Worker
  

Spark采用Executor的优点：

• 利用多线程来执行详细使命，淘汰使命的启动开销
  
• Executor中有一个BlockManager存储模块，会将内存和磁盘共同作为存储设备
  

Spark运行根本流程：Spark使命调度流程

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

• RDDObject是Spark中的基础数据结构，它包罗了数据的逻辑结构和物理结构。
  
• Stage是Spark中的使命级别，它包罗了一组Task，每个Task是独立的盘算单元。
  
• Task是Spark中的使命单元，它包罗了一组操作，每个操作是一个独立的盘算单元。
  
• 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摆设方式

四种不通范例的摆设方式：

• 单机：Local
  
• 集群：Standalone、Spark on Mesos、Spark on YARN
  

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

Spark的根本环境设置

• Linux
  
• Hadoop
  
• JDK
  
• Spark
  

环境变量

• JAVA_HOME
  
• HADOOP_HOME
  
• PYTHONE_HOME
  
• SPARK_HOME
  
• PYSPARKPATH
  
• PYSPARK_PYTHON
  
• 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的创建：文件、并行集合

• 当地加载创建
  lines = sc.textFile(“file:///usr/local/spark/1.txt”)
  lines.foreach(print)
  
• 分布式文件体系加载创建下面三选一即可
  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)
  
• 通过并行集合创建
  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 xx,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 xx,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 xx,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.评估模型拟合程度的准则和方法

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。