目录
一. Spark简介
二 . Spark 框架模块
三. 环境准备
3.1 Spark Local模式搭建
3.2 通过Anaconda安装python3环境
3.3 PySpark库安装
四 . Spark集群模式架构介绍
五. pycharm远程开发环境
六. Spark词频统计
一. Spark简介
1. Spark 和MapReduce
MR:大量的磁盘反复写入,运行是基于进程进行数据处理,创建和销毁的过程比较消耗资源,并且速度较慢
Spark:基于线程执行任务,引入新的数据结构RDD(弹性分布式数据集),让spark基于内存进行运行,比磁盘速度快很多
2. 进程与线程
进程:app,不同软件
线程:微信里的每个聊天对话
3. 四大特点
高效性 (快), 易用性(多语言), 通用性(提供了多个工具库), 兼容性(任何地方运行)
二 . Spark 框架模块
Spark Core:实现Spark的基本功能
Spark Sql : 使用Sql处理结构化数据
Strctured Streaming :基于Spark SQL进行流式/实时的处理组件,主要处理结构化数据
三. 环境准备
三台虚拟机 的快照恢复到spark阶段
主机做ip地址映射
3.1 Spark Local模式搭建
local模式 指的是在本地一个进程中,创建多个线程来模拟Spark程序分布式运行
1.把包放到虚拟机中
2. cd命令进到目录中,然后命令解压
tar -zxf spark-3.1.2-bin-hadoop3.2.tgz -C /export/server/
解压后重命名一下
mv spark-3.1.2-bin-hadoop3.2 spark
之后 进到bin目录
cd /export/server/spark/bin
使用命令开启spark
./spark-shell
Spark context available as 'sc' (master = local[*], app id = local-1704164639145).
Spark session available as 'spark'.其中''sc ''是SparkContext实例对象
''spark'' 是SparkSession 实例对象
3.2 通过Anaconda安装python3环境
Spark是一个独立的框架
PySpark是Python的库,由Spark官方提供
1.下载好包
2.上传到三台虚拟机中
3.执行脚本
bash Anaconda3-2021.05-Linux-x86_64.sh
运行后阅读一堆注意事项,然后yes,回车,开始下载库,会有点久,三台都需要运行
4.配置环境变量
配置anaconda的环境变量，三个节点都需要进行相同的操作:
vim /etc/profile
##增加如下配置
export ANACONDA_HOME=/root/anaconda3/bin
export PATH=$PATH:$ANACONDA_HOME
重新加载环境变量: source /etc/profile
修改bashrc文件，三个节点都需要进行相同的操作：
sudo vim ~/.bashrc
在最上面添加如下内容:
export PATH=~/anaconda3/bin:$PATH
添加完后,node1重新连接一下,如果命令前面出现了(base),就再去到 vim ~/.bashrc ,在最后一行插入conda deactivate
最后输入python,版本为3.8.8就是安装完了
3.3 PySpark库安装
1.把这个包传到虚拟机中
2.传好后不用解压,在software文件路径里直接敲命令运行
pip install pyspark-3.1.2.tar.gz
py4j: 将python代码转换成java代码
运行Spark
cd /export/server/spark/bin
./pyspark
测试圆周率
cd /export/server/spark/bin
./spark-submit /export/server/spark/examples/src/main/python/pi.py 100
四 . Spark集群模式架构介绍
一个主角色 Master的职责:
1.管理众多的从角色
2.负责资源管理和任务的分配
3.将Driver进程分配给到Worker进行运行
4.接收Spark任务的请求
多个从角色 Worker的职责 :
1. 负责具体任务的执行
2.向Master汇报心跳,汇报任务运行的状态
3.Driver随机选择一个Worker从角色进行启动和运行
4.接收执行来自Driver分配来的任务
Zookeeper高可用环境下为了防止单点故障,会有多个Master,主节点的角色分为Active和Standby
五. pycharm远程开发环境
远程连接方案, 允许所有的程序员都去连接远端的测试环境, 确保大家的环境都是统一的, 避免各种环境问题的发生,而且由于是连接的远程环境, 所有在pycharm编写代码 都会自动上传到远端服务器中, 在执行代码的时候, 相当于是直接在远端环境上进行执行操作
1.创建新项目是连接ssh
2.选择python解释器在虚拟机里的位置
3.pycharm里勾选tools的deployment 的automatic upload
4.pycharm里tools选择最下面的browse remote host,就可以在pycharm里显示虚拟机的文件目录了
六. Spark词频统计
1.需求描述
文本内容如下：hello hello sparkhello heima spark
代码流程步骤:
创建Spark Context对象
数据输入
数据处理
文件内容切分
数据格式转换
分组和聚合
数据输出
释放资源
需要自己建一个文件先,存储要统计的词语,路径为
'file:///export/data/2024.1.2_Spark/1.2_day01/content.txt'
import os
from pyspark import SparkConf, SparkContext
# 指定远端的环境地址
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
# pyspark程序运行需要有main函数入口
if __name__ == '__main__':
print("PySpark入门案例：WordCount词频统计")
# 创建SparkContext对象（Spark基础的顶级对象）
"""
setAppName：设置Spark程序运行时的名称
setMaster：设置Spark程序运行模式，这里是设置的本地模式
"""
conf = SparkConf().setAppName('spark_wordcount_demo').setMaster('local[*]')
sc = SparkContext(conf=conf)
# 数据输入
"""
textFile：读取文件。支持本地文件系统和HDFS文件系统
本地文件系统：file:///路径
HDFS文件系统：hdfs://node1:8020/路径
"""
init_rdd =sc.textFile('file:///export/data/2024.1.2_Spark/1.2_day01/content.txt')
# 数据处理
# 文本内容切分：flatMap
# 输出结果：['hello', 'hello', 'spark', 'hello', 'heima', 'spark']
flatmap_rdd = init_rdd.flatMap(lambda line: line.split(' '))
# 数据格式转换：map hello -> (hello,1)
# 输出结果：[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1),('spark', 1)]
map_rdd = flatmap_rdd.map(lambda word: (word,1))
# 分组和聚合：reduceByKey
# 输出结果：[('hello', 3), ('spark', 2), ('heima', 1)]
"""
reduceByKey：该算子是先对数据按照key进行分组，分组的时候会将value放到一个List列表当中。然
后对value形成的List列表进行聚合处理
lambda agg,curr: agg+curr 这个自定义的lambda表达式的实际含义是：agg是中间临时聚合结果，
默认值是列表中的第一个元素，
curr是当前遍历到的元素，默认值是列表中的第二个元素。agg+curr实际上是等同于agg=agg+curr
底层运行过程说明，以hello为例,首先是分组
输入的是：[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1),
('spark', 1)]
分组后的结果： hello [1,1,1]
spark [1,1]
heima [1]
2-接着是聚合操作
第一次聚合：agg的值是1，curr的值是1，聚合结果是1+1=2，并且会将结果赋值给agg，所以第一次聚合
之后agg的就变成了2
第二次聚合：agg的值是2，curr的值是1，聚合结果是2+1=3，并且会将结果赋值给agg。由于已经将
value形成的List列表中元素遍历完成了，所以最终hello的结果就是3
"""
result_rdd = map_rdd.reduceByKey(lambda agg, curr: agg + curr)
# 数据输出
print(result_rdd.collect())
# 释放资源
sc.stop()
结果:
