基于Jupyter平台通过python实现Spark的应用程序之wordCount

1、启动spark平台，界面如下：



2、启动Jupyter，界面如下图所示：



如果你对以上启动存在疑问的话，请看我的上一篇博客，关于Jupyter配置Spark的。

3、功能分析

- 我们要实现的一个功能是统计词频

- 我们需要把统计的文件上传到hdfs里面

- 编写python脚本

4、代码实现

- 上传文件到hdfs

我有一个hello.txt文件，里面有两行内容，如下图所示：



文件放到了~/Downloads下，进入hadoop/bin文件，输入如下命令：

./hadoop fs -put ~/Downloads/hello.txt

查看上传是否成功，输入如下命令：

./hadoop fs -ls /data |grep he*

*代表匹配he后边所有的内容，输出结果截图如下：



- 在Jupyter平台开启python编程页面，引入spark包，代码如下：

from pyspark import SparkContext, SparkConf

初始化sc，这里不了解spark编程原理的话建议看下spark官网，代码如下：

conf = SparkConf().setAppName("wordcount").setMaster("local")

sc = SparkContext(conf = conf)

setAppName设置的我这个运行实例的名字，setMaster设置的是运行的模式，这里选择本地模式。

- 读取本地文件，代码如下：

lines = sc.textFile("hdfs://hadoop0:9000/data/hello.txt")

- 使用spark的flatMap算子，将读入的文件拆成一个个单词，代码如下：

words = lines.flatMap(lambda line:line.split())

注意这个时候words是一个字符串数组，里面存的是hello，you，hello，me。这里用到了spark中提供transformations和pair两种RDD，前一种只是做一个标记不执行，后一种才是真正的执行，这里具体的运行流程我稍后会做一个详细的分析给大家，让大家彻底理解spark的运行原理。

- 使用spark的map算子，将words设置为(key,value) 形式，代码如下：

pairs = words.map(lambda word:(word,1))

注意这个时候pairs 存放的是一个字符串数组，内容是：(hello,1),(you,1),(hello,1),(me,1)

- 使用spark的reduceByKey算子，统计相同key出现的次数，即key相同的把value值相加，key合并为一个，代码如下：

counts = pairs.reduceByKey(lambda a,b:a + b)

注意这个时候counts为一个字符串数组，内容是：(hello,2),(you,1),(me,1)

- 使用spark的collect算子,将结果打印出来，代码如下：

counts.collect() ``

- 还可以使用spark的reduceBykey，对结果按key排序，代码如下：

counts.countByKey()#根据key排序“`

注意输出结果变成(hello,2),(me,1),(you,1),整个截图如下：