Apache Spark是用于大规模数据(large-scala data)处理的统一(unified)分析引擎.
mapreduce架构图如下:
MapReduce的主要缺点:
Spark相对MapReduce的优点:
高效性
易用性
通用性
兼容性(任何地方运行)
支持三方工具接入
支持多种操作系统
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文件系统和本地文件系统
HDFS文件系统:hdfs://node1:8020/文件路径
本地文件系统: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元素,默认取列表中的第二个元素
第一次聚合:
agg =列表中的第一个1,curr=列表中的第二个1。聚合结果agg+curr=1+1=2,再将2赋值给agg
第二次聚合:
agg =上次临时聚合结果2,curr=列表中的第三个1。聚合结果agg+curr=2+1=3,再将3赋值给agg
最后发现已经遍历到value列表的最后一个元素,因此聚合过程结果。最终的hello的次数,就是3
"""
result = map_rdd.reduceByKey(lambda agg,curr: agg+curr)
# 4- 数据输出
"""
collect():用来收集数据,返回值类型是List列表
"""
print(result.collect())
# 5- 释放资源
sc.stop()
可能出现的错误:
原因: 找不到JAVA_HOME环境
解决方案: 需要在代码中指定远端的环境地址 以及 在node1环境中初始化JAVA_HOME地址
第一步: 在mian函数上面添加以下内容
os.environ[‘SPARK_HOME’] = ‘/export/server/spark’
os.environ[‘PYSPARK_PYTHON’] = ‘/root/anaconda3/bin/python3’
os.environ[‘PYSPARK_DRIVER_PYTHON’] = ‘/root/anaconda3/bin/python3’
第二步:在node1的 /root/.bashrc 中配置初始化环境的配置
export JAVA_HOME=/export/server/jdk1.8.0_241
注意: 路径需要填写你自己