系列文章目录
第1天总结：spark基础学习
- 1- Spark基本介绍（了解）
- 2- Spark入门案例（掌握）
- 3- 常见面试题（掌握）
文章目录
前言
本文就介绍了spark学习的基础内容
以及详细介绍了词频统计案例。
一、Spark基本介绍
1、Spark是什么
1.1 定义
Apache Spark是用于大规模数据（large-scala data）处理的统一（unified）分析引擎。
1.2 Spark与MapReduce对比(面试题)
MapReduce架构回顾
- MapReduce的主要缺点：- 1- MapReduce是基于进程进行数据处理，进程相对线程来说，在创建和销毁的过程比较消耗资源，并且速度比较慢- 2- MapReduce运行的时候，中间有大量的磁盘IO过程。也就是磁盘数据到内存，内存到磁盘反复的读写过程- 3- MapReduce只提供了非常低级（底层）的编程API，如果想要开发比较复杂的程序，那么就需要编写大量的代码。
- Spark相对MapReduce的优点：- 1- Spark底层是基于线程来执行任务- 2- 引入了新的数据结构——RDD（弹性分布式数据集），能够让Spark程序主要基于内存进行运行。内存的读写数据相对磁盘来说，要快很多- 3- Spark提供了更加丰富的(顶层)编程API，能够非常轻松的实现功能开发
2、Spark特点
快速记忆: speed, easy use , general , runs everywhere
- 高效性- 计算速度快 - 提供了一个全新的数据结构RDD（弹性分布式数据集）。整个计算操作，基于内存计算。当内存不足的时候，可以放置到磁盘上。整个流程是基于DAG(有向无环图)执行方案。- Task线程完成计算任务执行
- 易用性- 支持多种语言开发 （Python，SQL，Java，Scala，R），降低了学习难度
- 通用性- 在 Spark 的基础上，Spark 还提供了包括Spark SQL、Spark Streaming、MLlib 及GraphX在内的**多个工具库(模块)**，我们可以在一个应用中无缝地使用这些工具库。
- 兼容性（任何地方运行）- 支持三方工具接入- 存储工具 - hdfs- kafka- hbase- 资源调度 - yarn- Kubernetes（K8s容器）- standalone（spark自带的）- 高可用 - zookeeper- 支持多种操作系统- Linux- windows- Mac
3、Spark框架模块
- Spark Core API：实现了 Spark 的基本功能。包含RDD、任务调度、内存管理、错误恢复、与存储系统交互等模块。数据结构RDD。–重点学习
- Spark SQL：我们可以使用 SQL处理结构化数据。数据结构：Dataset/DataFrame = RDD + Schema。–重点学习
- Structured Streaming：基于Spark SQL进行流式/实时的处理组件，主要处理结构化数据。–部分学习
- Streaming（Spark Streaming）：提供的对实时数据进行流式计算的组件，底层依然是离线计算，只不过时间粒度很小，攒批。–了解
- MLlib：提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等。–了解
- GraphX：Spark中用于图计算的API，性能良好，拥有丰富的功能和运算符，能在海量数据上自如地运行复杂的图算法。–了解
二、Spark入门案例（掌握）
1、需求描述
读取文本文件，文件内容是一行一行的文本，每行文本含有多个单词，单词间使用空格分隔。统计文本中每个单词出现的总次数。WordCount词频统计。
文本内容如下：
hello hello spark
hello heima spark
2、需求分析
Python编程思维的实现过程：
PySpark实现过程：
编程过程总结：
- 1.创建SparkContext对象
- 2.数据输入
- 3.数据处理- 3.1文本内容切分- 3.2数据格式转换- 3.3分组和聚合
- 4.数据输出
- 5.释放资源
3、代码编写
可能出现的错误:
结果: 可能会报错: JAVA_HOME is not set
原因: 找不到JAVA_HOME环境
解决方案: 需要在代码中指定远端的环境地址 以及 在node1环境中初始化JAVA_HOME地址
第一步:在node1的 /root/.bashrc 中配置初始化环境的配置
vim /root/.bashrc
export JAVA_HOME=/export/server/jdk1.8.0_241
第二步: 在main函数上面添加以下内容
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
注意: jdk路径配置到node1的 /root/.bashrc 文件的第三行,示例如下:
代码：
from pyspark import SparkConf, SparkContext
import os
绑定指定的Python解释器
os.environ[‘SPARK_HOME’] = ‘/export/server/spark’
os.environ[‘PYSPARK_PYTHON’] = ‘/root/anaconda3/bin/python3’
os.environ[‘PYSPARK_DRIVER_PYTHON’] = ‘/root/anaconda3/bin/python3’
创建main函数
if name == ‘main’:
print(“Spark入门案例: WordCount词频统计”)
# 1- 创建SparkContext对象
"""
setAppName：设置PySpark程序运行时的名称
setMaster：设置PySpark程序运行时的集群模式
"""
conf = SparkConf()\
.setAppName('spark_wordcount_demo')\
.setMaster('local[*]')
sc = SparkContext(conf=conf)
# 2- 数据输入
"""
textFile：支持读取HDFS文件系统和linux本地文件系统
HDFS文件系统：hdfs://node1:8020/文件路径
linux本地文件系统：file:///文件路径
"""
init_rdd = sc.textFile("file:///export/data/gz16_pyspark/01_spark_core/data/content.txt")
# 3- 数据处理
# 文本内容切分
"""
flatMap运行结果：
输入数据：['hello hello spark', 'hello heima spark']
输出数据：['hello', 'hello', 'spark', 'hello', 'heima', 'spark']
map运行结果：
输入数据：['hello hello spark', 'hello heima spark']
输出数据：[['hello', 'hello', 'spark'], ['hello', 'heima', 'spark']]
"""
# flatmap_rdd = init_rdd.map(lambda line: line.split(" "))
flatmap_rdd = init_rdd.flatMap(lambda line: line.split(" "))
# 数据格式转换
"""
输入数据：['hello', 'hello', 'spark', 'hello', 'heima', 'spark']
输出数据：[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
"""
map_rdd = flatmap_rdd.map(lambda word: (word,1))
# 分组和聚合
"""
输入数据：[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
输出数据：[('hello', 3), ('spark', 2), ('heima', 1)]
reduceByKey底层运行过程分析：
1- 该算子同时具备分组和聚合的功能。而且是先对数据按照key进行分组，对相同key的value会形成得到List列表。再对分组后的value列表进行聚合。
2- 分组和聚合功能不能分割，也就是一个整体
结合案例进行详细分析：
1- 分组
输入数据：[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
分组后的结果：
key value列表
hello [1,1,1]
spark [1,1]
heima [1]
2- 聚合（以hello为例）
lambda agg,curr: agg+curr -> agg表示中间临时value聚合结果，默认取列表中的第一个元素；curr表示当前遍历到的value元素，默认取列表中的第二个元素
最后发现已经遍历到value列表的最后一个元素，因此聚合过程结果。最终的hello的次数，就是3
"""
result = map_rdd.reduceByKey(lambda agg,curr: agg+curr)
# 4- 数据输出
"""
collect()：用来收集数据，返回值类型是List列表
"""
print(result.collect())
# 5- 释放资源
sc.stop()
###运行结果：
总结
常见面试题
1.spark和mr的区别
Spark和MR（通常指的是Hadoop MapReduce）在多个方面存在显著的区别。
Spark在计算速度、并行度、资源利用率、编程灵活性和资源申请与释放等方面，相较于Hadoop MapReduce具有显著的优势。这使得Spark在处理大规模数据集和分析任务时，成为了一个更加高效和灵活的选择。
以下是它们之间的主要差异：
计算速度与迭代计算：
Spark：除了需要shuffle的计算外，Spark将结果/中间结果持久化到内存中，因此避免了频繁的磁盘I/O操作。这使得Spark在处理需要频繁读写中间结果的迭代计算时，比MR具有更高的效率。
MR：所有的中间结果都需要写入磁盘，并在下一个阶段从磁盘中读取，这导致了较高的磁盘I/O开销和较低的计算速度。
并行度与任务调度：
Spark：将不同的计算环节抽象为Stage，允许多个Stage既可以串行执行，又可以并行执行。这种基于DAG（有向无环图）的任务调度执行机制，提高了任务的并行度和整体执行效率。
MR：任务之间的衔接涉及I/O开销，且下个任务的执行依赖于上个任务的结果，这限制了其并行度和处理复杂、多阶段计算任务的能力。
资源模型：
Spark：基于线程，采用多进程多线程模型。在同一个节点上，多个任务可以共享内存和资源，提高了数据和资源的利用率。
MR：基于进程，采用多进程单线程模型。每个任务都是独立的进程，申请资源和数据都是独立进行的，这导致了较高的资源申请和释放开销。
编程模型与灵活性：
Spark：提供了多种数据集操作类型，包括转换算子、行动算子和持久化算子，使得编程模型比Hadoop MapReduce更灵活。同时，Spark支持使用Scala、Java、Python和R语言进行编程，具有更好的易用性。
MR：只有map和reduce两个类，相当于Spark中的两个算子，其编程模型相对较为简单和固定。
资源申请与释放：
Spark：多个task运行在同一个进程中，这个进程会伴随Spark应用程序的整个生命周期。即使在没有作业进行时，进程也是存在的，这避免了频繁的进程创建和销毁开销。
MR：每个task都是一个独立的进程，当task完成时，进程也会结束。这导致了较高的进程创建和销毁开销。
综上所述，
2.Spark的四大特性：
Speed（高速性）：
Spark是一个基于内存计算的分布式计算框架，能够在内存中直接处理数据，减少了磁盘I/O的开销，从而显著提高了计算速度。
官方数据表明，如果数据从内存中读取，Spark的速度可以高达Hadoop MapReduce的100多倍；即使数据从磁盘读取，Spark的速度也是Hadoop MapReduce的10倍以上。
Spark通过DAG（有向无环图）执行引擎支持无环数据流，使得数据处理更加高效。
Ease of Use（易用性）：
Spark提供了丰富的API，支持多种编程语言，如Scala、Java、Python和R，使得用户可以轻松地开发复杂的分布式应用程序。
Spark的易用性还体现在其支持的高级功能上，如SQL查询、机器学习和图计算等，这些功能都通过简洁的代码接口提供。
Generality（通用性）：
Spark生态圈即BDAS（伯克利数据分析栈）包含了多个组件，如Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等，这些组件能够无缝集成并提供一站式解决平台。
Spark Core提供内存计算框架，Spark SQL支持即席查询，Spark Streaming处理实时数据流，MLlib和MLbase支持机器学习，GraphX则专注于图处理。
Runs Everywhere（随处运行）：
Spark具有很强的适应性，能够读取多种数据源，如HDFS、Cassandra、HBase、S3和Techyon等。
Spark支持多种部署模式，包括Hadoop YARN、Apache Mesos、Standalone（独立部署）以及云环境（如Kubernetes）等，使得用户可以根据自身需求选择合适的部署方式。
综上所述，Spark的四大特性包括高速性、易用性、通用性和随处运行，这些特性使得Spark在处理大规模数据集和分析任务时表现出色，成为大数据处理领域的重要工具。
3.spark为什么执行快
Spark执行速度快的原因：
Spark通过内存计算、DAG执行引擎、RDD、任务调度优化、容错性和分布式计算等特性，实现了高性能的数据处理能力，从而能够在处理大规模数据集和分析任务时表现出色。
内存计算：
Spark采用了内存计算的方式，将数据和中间计算结果存储在内存中，而不是传统的硬盘中。
由于内存的速度远快于硬盘，因此Spark能够避免频繁的磁盘I/O操作，从而显著提高了数据处理的速度。
官方数据表明，如果数据从内存中读取，Spark的速度可以高达Hadoop MapReduce的100多倍；即使数据从磁盘读取，Spark的速度也是Hadoop MapReduce的10倍以上。
DAG执行引擎：
Spark采用了基于有向无环图（DAG）的执行引擎，将作业转化为一系列的有向无环图进行计算。
DAG执行引擎可以优化任务调度和计算，使得多个任务能够并行执行，进一步提高了计算效率。
弹性分布式数据集（RDD）：
RDD是Spark的核心数据模型，提供了对数据集的高效分布式处理。
RDD具有不可变性，但可以通过一系列的转换操作生成新的RDD，并支持在内存中缓存RDD，从而提高计算性能。
任务调度优化：
Spark将用户的代码转化为一系列的任务，并以有向无环图（DAG）的形式进行调度执行。
Spark的任务调度器可以根据数据的依赖关系来优化任务的执行顺序，将多个相关的任务合并在一起执行，减少了任务调度的开销。
容错性：
Spark通过将数据划分成多个分区，并在集群中复制多份数据来实现容错性。
当某个计算节点发生故障时，Spark可以自动将计算任务转移到其他节点上，并重新执行失败的任务，确保了计算的完整性和准确性。
分布式计算：
Spark支持分布式计算，能够将数据分成多个分区，并分布到不同的计算节点上进行并行处理。
这种分布式计算的方式能够充分利用集群资源，提高计算效率。
4.Spark词频统计的步骤以及每步涉及到的算子作用：
步骤一：基于文本文件创建RDD
使用sc.textFile(“/path/to/file.txt”)读取文本文件，并创建一个RDD（弹性分布式数据集）。
涉及到的算子：无。这是数据输入步骤，不涉及Spark的转换或行动算子。
步骤二：按空格拆分作扁平化映射
使用flatMap(_.split(" "))将RDD中的每一行文本按空格拆分成单词，并将所有单词合并成一个新的RDD。
涉及到的算子：flatMap。这是一个转换算子（Transformation），它会对RDD中的每个元素应用一个函数，并返回一个新的RDD，其中包含所有函数输出的元素。
步骤三：将单词数组映射成二元组数组
使用map((_, 1))将每个单词映射为一个二元组（单词，1），表示该单词出现了一次。
涉及到的算子：map。这也是一个转换算子，它将RDD中的每个元素转换成一个新的元素。
步骤四：将二元组数组按键归约
使用reduceByKey(_ + _)对二元组RDD进行归约操作，将具有相同键（即单词）的二元组合并，并将它们的值（即计数）相加。
涉及到的算子：reduceByKey。这是一个转换算子，它会对具有相同键的元素进行归约操作，并返回一个新的RDD。
步骤五：将词频统计结果按次数降序排列
使用sortBy(_._2, false)对词频统计结果进行排序，按照单词出现的次数从高到低排序。
涉及到的算子：sortBy。这是一个转换算子，它会对RDD中的元素进行排序，并返回一个新的RDD。
步骤六（可选）：收集并输出结果
使用collect将排序后的词频统计结果收集到驱动程序节点，并使用foreach(println)输出结果。
涉及到的算子：collect和foreach。collect是一个行动算子（Action），它会触发Spark作业的执行，并将RDD中的所有元素收集到驱动程序节点。foreach是一个行动算子，它会对RDD中的每个元素应用一个函数，但该函数不返回任何值。
案例总结：
在上述步骤中，flatMap、map、reduceByKey和sortBy是转换算子，它们用于创建和转换RDD；而collect和foreach是行动算子，它们会触发Spark作业的执行，并返回结果或进行其他操作。这些算子的组合使用，使得Spark能够高效地进行词频统计任务。
版权归原作者 大数据小朋友 所有， 如有侵权，请联系我们删除。