待我學有所成，結髮與蕊可好。@夏瑾墨

一個早上只做了一點微小的工作，很懺愧。但是發現Spark這玩意還是蠻有意思的。下面給大家介紹一下如何用python跑一遍Wordcount的詞頻統計的示例程式。

#在pyspark模組中引入SparkContext和SparkConf類
#在operator模組中匯入add類
from pyspark import SparkContext, SparkConf 
from operator import add

#應用程式名
#初始化一個SparkContext，現在sc就是一個SparkContext的例項化物件，然後方可建立RDD。
 

appName = "WordCount"
conf = SparkConf().setAppName(appName).setMaster("local")
sc = SparkContext(conf=conf)

# inputFiles表示輸入檔案路徑
# stopWordFile表示停詞檔案路徑
# outputFile表示輸出檔案路徑
inputFiles = "/home/hadoop/software/spark-2.0.0-bin-hadoop2.6/examples/src/main/resources/wordcount/*"
stopWordFile = "/home/hadoop/software/spark-2.0.0-bin-hadoop2.6/examples/src/main/resources/wordcount/stopword.txt"
 

outputFile = "/tmp/result"

#處理非單詞符號
targetList = list('\t\().,?[]!;|') + ['--']
#用空格替換這些標點符號，同時將替換後的行拆分成單詞.在flatMap中使用replaceAndSplit函式
def replaceAndSplit(s):
    for c in targetList:
        s = s.replace(c, " ")
    return s.split()

inputRDD = sc.textFile(inputFiles)
stopRDD = sc.textFile(stopWordFile)
stopList = stopRDD.map(lambda
 
 x: x.strip()).collect()

inputRDDv1 = inputRDD.flatMap(replaceAndSplit)
inputRDDv2 = inputRDDv1.filter(lambda x: x not in stopList)
inputRDDv3 = inputRDDv2.map(lambda x: (x,1))
inputRDDv4 = inputRDDv3.reduceByKey(add)
inputRDDv5 = inputRDDv4.map(lambda x: (x[1], x[0]))
inputRDDv6 = inputRDDv5.sortByKey(ascending=False)
inputRDDv7 = inputRDDv6.map(lambda x: (x[1], x[0])).keys()
top100 = inputRDDv7.take(100)
result = sc.parallelize(top100)
result.saveAsTextFile(outputFile)

背景知識

1.任何Spark程式的編寫都是從SparkContext（或用Java編寫時的JavaSparkContext）開始的，SparkContext的初始化需要一個SparkConf物件，Sparkconf包括了Spark叢集配置的各種引數（比如主節點的URL）。初始化後，就可以用SparkContext物件所包含的各種方法來建立，操作分散式資料集和共享變數。

2.涉及的函式

• Python split()方法：通過指定分隔符對字串進行切片，如果引數num 有指定值，則僅分隔 num 個子字串。
• Python strip() 方法：用於移除字串頭尾指定的字元（預設為空格）。
• Python lambda()方法：用來建立匿名函式，lambda的主體是一個表示式，用來封轉有限的邏輯進去。
• Python內建的filter()函式 : 用於過濾序列,filter()也接收一個函式和一個序列.
• map( )方法:接收一個函式，應用到RDD中的每個元素，然後為每一條輸入返回一個物件。根據提供的函式對指定序列做對映。
• flatMap( )方法：接收一個函式replaceAndSplit，應用到RDD中的每個元素，返回一個包含可迭代的型別(如list等)的RDD,可以理解為先Map()，後flat().
-

map函式會對每一條輸入進行指定的操作，然後為每一條輸入返回一個物件；而flatMap函式則是兩個操作的集合——正是“先對映後扁平化”：
操作1：同map函式一樣：對每一條輸入進行指定的操作，然後為每一條輸入返回一個物件
操作2：最後將所有物件合併為一個物件

• Spark sortByKey函式 : 作用於Key-Value形式的RDD，並對Key進行排序。它是在org.apache.spark.rdd.OrderedRDDFunctions中實現的.
• take(): Spark的RDD的action操作take()用於提取資料
• parallelize() : 建立一個並行集合,例如sc.parallelize(0 until numMappers, numMappers) 建立並行集合的一個重要引數，是slices的數目（例子中是numMappers），它指定了將資料集切分為幾份.
• Spark主要提供了兩種函式：parallelize和makeRDD：
  1）parallelize的宣告：

def parallelize[T: ClassTag](    
 seq: Seq[T],    
numSlices: Int = defaultParallelism): RDD[T]   

2）makeRDD的宣告：

def makeRDD[T: ClassTag](    

seq: Seq[T],    
 numSlices: Int = defaultParallelism): RDD[T]    
def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])]): RDD[T]   

3）區別：

A）makeRDD函式比parallelize函式多提供了資料的位置資訊。
B）兩者的返回值都是ParallelCollectionRDD，但parallelize函式可以自己指定分割槽的數量，而makeRDD函式固定為seq引數的size大小。

![這裡寫圖片描述](https://img-blog.csdn.net/20161023140214492)

使用spark-submit執行python檔案，我們選擇使用local模式

以下是詞頻統計結果：

參考資料

待我學有所成，結髮與蕊可好。@夏瑾墨