文本数据操作的实用演示示例

本实例是实时统计本地文本文件的单词个数，在/root/sparkTest/test目录下每隔21s统计一下新添加的文件中各个以空格分割的单词的个数。此案例常用来以准实时的方式对互联网网站的流量进行不同粒度的统计。步骤如下：

（1）建立名称为“SparkStreamingExample”的项目工程，如图7-8所示。

图7-8 建立项目工程

（2）在WordCount.scala文件中具体实现实时统计本地文本文件的单词个数的功能。

（3）接下来对上述代码进行解析。

1）首先初始化并构建了一个SparkConf类的对象sparkConf，使用其中的setMaster和setAppName方法将这个驱动程序的Master设为本地模式，将其名称设为WordCount。然后使用这个SparkConf类对象初始化了一个StreamingContext类，将之前的配置项传递给spark_ ssc，这样就可以控制驱动程序了。setMaster（＂local[3]＂）表示驱动程序为本地模式的同时还表示开启了三个线程，对应sparkStreaming的操作一般都需要开启大于两个的线程，因为必须要有一个线程监听数据流，还有有其他的线程进行数据流的处理操作。

2）StreamingContext（sparkConf，Seconds（21））的第二个参数Seconds（21）表示每隔21s对数据流进行切割一次。

3）spark_ssc.textFileStream（＂/root/sparkTest/test＂）表示从本地文件系统的/root/spark-Test/test目录下读取文件，一旦这个目录下有新的文件进来的时候StreamingContext的实例对象spark_ssc就会读入这个文件并生成名称为sparkLines的RDD。sparkLines对文本中的每一行使用空格对单词进行切分sparkLines.flatMap（_.split（＂＂）），切分了之后就开始进行单词统计sparkWords.map（x＝＞（x，1））.reduceByKey（_+_）。最后把每一个时间段内增加的单词统计完成之后打印出来sparkWordCounts.print（）。

4）spark_ssc.awaitTermination（）表示等待数据流的终结，sparkStreaming会一直处于等待监听的状态。

（4）打开/root/sparkTest/test目录查看统计结果，如图7-9所示。(www.xing528.com)

此时发现在/root/sparkTest/test目录没有任何文件，这时在IntelliJ IDEA里运行Word-Count.scala文件来测试是否会有单词计数后的结果值。运行结果如下：

图7-9 /root/sparkTest/test目录

因为在/root/sparkTest/test目录下没有任何文件，所以没有统计到任何单词。现在我们在/root/sparkTest/test目录新建一个testSparkStreamingWordCount.txt文件，如图7-10所示。

testSparkStreamingWordCount.txt文件里分别有五个字母A、五个字母B、五个字母C、五个字母D，各自都使用空格分开。在IntelliJ IDEA里运行WordCount.scalad的结果如下：

图7-10 新建一个testSparkStreamingWordCount.txt文件

可以看到现在已经统计出了在/root/sparkTest/test目录新加入的testSparkStreamingWord-Count.txt文件中的各个字母的个数：A、B、C、D分别有5个。