1. 当前环境
pyspark：版本号为3.1.2
JAVA_JDK: 版本号为1.8.0_333
Hadoop: 版本号为3.3.0
2. 相关信息
2.1 相关文件
words.txt如下
hello world hello hadoop
hadoop hello world hive
hive hive hadoop
2.2 相关流程
3. 运行代码
# spark入门案例 --- WordCount
from pyspark import SparkContext,SparkConf
import findspark
findspark.init()
if __name__ == '__main__':
print("spark入门案例 --- WordCount")
# 1) 创建 sparkContext对象
conf = SparkConf().setMaster("local[*]").setAppName("WordCount")
# 自动返回 变量: ctrl +atl + v
sc = SparkContext(conf=conf)
# 2) 读取文件数据 file是调用本地文件
rdd_init = sc.textFile("file:///F:\python\学习\Spark/file/words.txt")
# 3) 对数据执行切割操作: 得到 ['hello', 'world', 'hello', 'hadoop', 'hadoop', 'hello', 'world', 'hive', 'hive', 'hive', 'hadoop']
rdd_flatMap = rdd_init.flatMap(lambda line:line.split(' ')) # 一个对多个flatMap
# 4) 对数据转换为 单词,1 操作
# [('hello', 1), ('world', 1), ('hello', 1), ('hadoop', 1), ('hadoop', 1), ('hello', 1), ('world', 1), ('hive', 1), ('hive', 1), ('hive', 1), ('hadoop', 1)]
rdd_map = rdd_flatMap.map(lambda word: (word,1)) # 一个对一个 map
# 5,6) 对数据执行分组操作 统计求和操作 groupByKey是指根据key完成自动分组 reduceByKey是指根据key来做聚合
# 从Shuffle的角度：
# groupByKey和reduceByKey都存在shuffle的操作，但是reduceByKey可以在shuffle之前对分区内相同key的数据集进行预聚合
# （combine），这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高。
# 从功能的角度：
# reduceByKey其实包含分组和聚合的功能；groupByKey只能分组，不能聚合，所以在分组聚合的场合下，
# 推荐使用reduceByKey，如果仅仅是分组而不需要聚合，那么还是只能使用groupByKey。
rdd_res = rdd_map.reduceByKey(lambda agg, curr: agg + curr)
# 7) 输出: 打印
print(rdd_res.collect())
# 8) 关闭 sparkContext对象
sc.stop()
4. 运行结果
5. 运行错误情况
5.1 py4j.protocol.Py4JJavaError
错误截图
错误原因
因为转载时间过长找不到spark所以报错
解决办法
import findspark
findspark.init()
5.2 JAVA_HOME is not set
出现位置: 当pycharm采用SSH连接远程Python环境时, 启动执行spark程序可能报出原因: 加载不到jdk的位置
解决方案:
- 第一步: 可以在linux的 /root/.bashrc 文件中, 添加以下两行内容 (注意需要三台都添加)
export JAVA_HOME=/export/server/jdk1.8.0_241
export PYSPARK_PYTHON=/root/anaconda3/bin/python
- 第二步： 在代码中, 指定linux中spark所在目录, spark中配置文件, 即可自动加载到: 锁定远端操作环境, 避免存在多个版本环境的问题
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ["PYSPARK_PYTHON"]="/root/anaconda3/bin/python"
os.environ["PYSPARK_DRIVER_PYTHON"]="/root/anaconda3/bin/python"
