spark练习例子——单词计数——pyspark
spark练习例子——单词计数——pyspark
- 练习科目一 ： 单词计数
- 引入依赖
- 启动sparkSession
- 读取文本，和单词
- 单词计数
- 总结
练习科目一 ： 单词计数
引入依赖
import sys
import os
os.environ["PYSPARK_PYTHON"] = "E:/code-private/python/venvs/venv38/Scripts/python.exe"
# os.environ["PYSPARK_DRIVER_PYTHON"] = "E:/code-private/python/venvs/venv38/Scripts/python.exe"
os.environ["JAVA_HOME"] = "D:/Program Files/Java/jdk-11.0.7"
os.environ["SPARK_HOME"] = "D:/Program/spark-3.2.0-bin-hadoop3.2-scala2.13"
os.environ["PYLIB"] = os.environ["SPARK_HOME"] + "/python/lib"
sys.path.insert(0, os.environ["PYLIB"] + "/py4j-0.10.7-src.zip")
sys.path.insert(0, os.environ["PYLIB"] + "/pyspark.zip")
我用的是pycharm 跑的 notebook ,
notice: 这里遇到问题就是，我指定了pyspark 是虚拟环境的python 3.8 ， 我 anaconda base 环境是3.7
跑pyspark ,一直都用的是 base 环境，一直都会报错，版本不兼容。
启动sparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("test").config("spark.network.timeout", "10000000").master(
"local[*]").getOrCreate()
spark.version
spark 2.13 与之前的练习版本有着不同差异，需要注意
读取文本，和单词
text = spark.read.text("in/word_count.text")
text.take(1)
words = text.rdd.flatMap(lambda s : s.value.split(" "))
words.take(1)
由其是这里，在dataframe 里面，必须要通过value 来取值，而不是 直接 s.split(" ") 去切割
单词计数
wordCounts = words.countByValue()
for word,count in wordCounts.items():
print("{} : {}".format(word,count))
总结
简单明了，无难度，需要适应新版本改变