数据准备

数据下载：《莎士比亚全集》

我们先来看看原始数据：首先将数据加载到RDD，然后显示数据框的前15行。

shakespeareDF = sqlContext.read.text(fileName)
shakespeareDF.show(15, truncate=False)

输出如下：

+-------------------------------------------------------+
|value                                                  |
+-------------------------------------------------------+
|1609                                                   |
|                                                       |
|THE SONNETS                                            |
|                                                       |
|by William Shakespeare                                 |
|                                                       |
|                                                       |
|                                                       |
|                     1                                 |
|  From fairest creatures we desire increase,           |
|  That thereby beauty's rose might never die,          |
|  But as the riper should by time decease,             |
|  His tender heir might bear his memory:               |
|  But thou contracted to thine own bright eyes,        |
|  Feed'st thy light's flame with self-substantial fuel,|
+-------------------------------------------------------+

数据清洗

因为原始数据包括标点符号，大小写字符，空行。所以我们需要对数据进行清洗。所以我提供了一个removePunctuation函数。这个函数将去掉了标点，删除了句子两端的多余的空格，并将字符全部转换为小写。

from pyspark.sql.functions import regexp_replace, trim, lower
def removePunctuation(column):
    return lower(trim(regexp_replace(column, '[^\w\s]', '')))

为了使用这个函数，我们先来看一个例子。

sentenceDF = (sqlContext
              .createDataFrame([('Hi, you!',),
                                (' No under_score!',),
                                (' *      Remove punctuation then spaces  * ',)], ['sentence']))
sentenceDF.show(truncate=False)

原始的数据框输出如下：

+------------------------------------------+
|sentence                                  |
+------------------------------------------+
|Hi, you!                                  |
| No under_score!                          |
| *      Remove punctuation then spaces  * |
+------------------------------------------+

接下来使用removePunctuation进行清洗。

from pyspark.sql.functions import col
(sentenceDF
 .select(removePunctuation(col('sentence')).alias('sentence'))
 .show(truncate=False))

清洗后的数据框输出如下：

+------------------------------+
|sentence                      |
+------------------------------+
|hi you                        |
|no under_score                |
|remove punctuation then spaces|
+------------------------------+

有了这个函数，我们就能对《莎士比亚全集》进行清洗了，首先将shakespeare.txt加载到RDD，并使用removePunctuation函数对数据进行清洗.

from pyspark.sql.functions import col
fileName = "shakespeare.txt"
shakespeareDF = (sqlContext
                 .read
                 .text(fileName)
                 .select(removePunctuation(col('value')).alias('value')))
shakespeareDF.show(15, truncate=False)

清洗后的数据框输出如下：

+-------------------------------------------------+
|value                                            |
+-------------------------------------------------+
|1609                                             |
|                                                 |
|the sonnets                                      |
|                                                 |
|by william shakespeare                           |
|                                                 |
|                                                 |
|                                                 |
|1                                                |
|from fairest creatures we desire increase        |
|that thereby beautys rose might never die        |
|but as the riper should by time decease          |
|his tender heir might bear his memory            |
|but thou contracted to thine own bright eyes     |
|feedst thy lights flame with selfsubstantial fuel|
+-------------------------------------------------+

接下来，我们使用split函数分隔每一行的句子，然后用explode函数将行转列，得到一个包括所有单词的数据框，最后使用where函数过滤掉数据框的空行。

from pyspark.sql.functions import split, explode
shakeWordsDF = (shakespeareDF
                .select(explode(split(shakespeareDF.value, ' ')).alias('word'))
                .where("word<>''"))
shakeWordsDF.show()
shakeWordsDFCount = shakeWordsDF.count()
print shakeWordsDFCount

转换后的数据框输出如下:

+-----------+
|       word|
+-----------+
|       1609|
|        the|
|    sonnets|
|         by|
|    william|
|shakespeare|
|          1|
|       from|
|    fairest|
|  creatures|
|         we|
|     desire|
|   increase|
|       that|
|    thereby|
|    beautys|
|       rose|
|      might|
|      never|
|        die|
+-----------+

数据统计

为了统计单词数，我提供一个wordCount函数，它作用是按单词进行分组，然后统计各个分组中单词的个数，最后返回包含word和count列的数据框。

def wordCount(wordListDF):
    return wordListDF.groupBy('word').count()

先来看一个使用wordCount函数的例子：

wordsDF = (sqlContext
           .createDataFrame([('cat',), ('elephant',), ('rat',), ('rat',), ('cat', )], ['word']))
wordCount(wordsDF).show()
wordCount(words)

wordCount函数返回的数据框输出如下：

+--------+-----+
|    word|count|
+--------+-----+
|     cat|    2|
|     rat|    2|
|elephant|    1|
+--------+-----+

接下来使用wordCount函数统计《莎士比亚全集》的单词数，然后按照count列降序排列。

from pyspark.sql.functions import desc
topWordsAndCountsDF = wordCount(shakeWordsDF).orderBy(desc('count'))
topWordsAndCountsDF.show()

排序后的数据框输出如下所示

+----+-----+
|word|count|
+----+-----+
| the|27361|
| and|26028|
|   i|20681|
|  to|19150|
|  of|17463|
|   a|14593|
| you|13615|
|  my|12481|
|  in|10956|
|that|10890|
|  is| 9134|
| not| 8497|
|with| 7771|
|  me| 7769|
|  it| 7678|
| for| 7558|
|  be| 6857|
| his| 6857|
|your| 6655|
|this| 6602|
+----+-----+

总结

可以看到，出现次数较多的单词大都是停用词。