Spark サンプルアプリケーション#1 WordCount and Word Ranking
WordCount
Sparkのパッケージに含まれているサンプルコードを基に何を行っているか確認する。
from __future__ import print_function
import sys
from operator import add
from pyspark import SparkContext
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: wordcount <file>", file=sys.stderr)
exit(-1)
sc = SparkContext(appName="PythonWordCount")
lines = sc.textFile(sys.argv[1], 1)
counts = lines.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1).reduceByKey(add)
output = counts.collect()
for (word, count) in output:
print("%s: %i" % (word, count))
sc.stop()
アプリを開発する際、データがどのように格納・処理されているか確認しながら進めていった方がよいので、PySparkを使って上記のサンプルコードを確認していく。
-SparkContextの生成
sc = SparkContext(appName="PythonWordCount")
SparkContextは、Sparkのアプリ全体の実行関する情報を集約したもの。Sparクラスタへの接続を行ったり、RDDを作成するために使用されたり、アキュムレートやブロードキャストなどの共用変数を提供する。
- テキストファイルからRDDを生成
textRDD = sc.textFile("/opt/spark/README.md")
textFileメソッドを使用し、テキストファイルからRDDを生成する。
※RDDのパーティションサイズはファイルサイズを128MBで割った数近くになる。
※RDDの要素をフィルタリングしたい場合は、filterメソッドを使用する。filterメソッドは、RDDに含まれる要素にたいしてフィルタリング条件にマッチする要素のみを残したRDDを生成する。
ex) wordRDD = textRDD.filter(labda x: x.isalnum()) //英数字だけを取得
- RDDの要素を加工
lines.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1)
x.splitでwordに分解し、flatMapで各行のwordをflatten化して一つの大きな配列にする。mapメソッドにより、各要素を変換する。このプログラムではタプル(x,1)に変換。
- RDDの要素をキー毎に集約
counts = lines.flatMap(lambda x: x.split(' ')).map(lambda x: (x, 1).reduceByKey(add)
reduceByKeyメソッドにより、RDDに含まれる要素を、同じキーを持つものにグループ分けし、グループ毎に値を集約する。
reduceByKeyメソッドが対象としている要素のデータ型は(Key,Value)。
reduceByKeyメソッドではキー毎の集約処理にシャッフルを行っているので、エグゼキュータ間の多対多の通信が発生する。
- RDDの要素をキー毎に集約
output = counts.collect()
for (word, count) in output:
print("%s: %i" % (word, count))
collectメソッドでRDDの要素を取り出し、printで出力する。
- アプリケーションの停止
sc.stop()
stopメソッドを呼び出しアプリを停止する。
Appendix) WordCount ranking
どのwordが多く出現しているのかトップ10ランキングを表示する。
sortByKeyメソッドを利用することによりRDDに含まれる要素をソートすることができる。
(Key,Value)を要素に持つRDDが対象で、Keyによる大小に基づいてsortする。
そのためKeyが文字のままだとsortがアルファベット順になるので、(word,count)から(count,word)に変更する必要がある。
実際にランキングを表示してみたところ記号とかも入ってたので記号などは除外するようにした。
from __future__ import print_function
import sys
from operator import add
from pyspark import SparkContext
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: wordcount <file>", file=sys.stderr)
exit(-1)
sc = SparkContext(appName="PythonWordCount")
lines = sc.textFile(sys.argv[1], 1)
counts = lines.flatMap(lambda x: x.split(' '))\
.filter(lambda x: x.isalnum())\
.map(lambda x: (x, 1))\
.reduceByKey(add)
wordrankingRDD = counts.map(lambda (x,y): (y,x))
sortedwordrankingRDD = wordrankingRDD.sortByKey(ascending=False)
output = sortedwordrankingRDD.take(10) #要素を先頭から10個取り出す
for (count, word) in output:
print("%s: %i" % (word, count))
sc.stop()
$ ./bin/spark-submit --master=local examples/src/main/python/wordranking.py /tmp/README.md
the: 21
to: 14
Spark: 14
for: 12
and: 10
a: 10
run: 7
on: 6
can: 6
is: 6
Author And Source
この問題について(Spark サンプルアプリケーション#1 WordCount and Word Ranking), 我々は、より多くの情報をここで見つけました https://qiita.com/t-yotsu/items/2dd8cfb932d424aee3d2著者帰属:元の著者の情報は、元のURLに含まれています。著作権は原作者に属する。
Content is automatically searched and collected through network algorithms . If there is a violation . Please contact us . We will adjust (correct author information ,or delete content ) as soon as possible .