Spark Streamingの動作確認 ローカル及びクラスタ環境

Spark入門の７章及び以下のサイトを参考にSparkStreamingの動作確認を行なった際のメモ。
http://spark.apache.org/docs/latest/streaming-programming-guide.html

ローカル環境でSpark Streamingの動作確認

ポート9999番で待ち受けるデータサーバ（データを書き込む）がある。このネットワークソケットに接続して、1秒毎にテキストデータをストリームデータとして読み込み、そのストリームデータに含まれる単語数をカウントするプログラムの動作確認を行う。
ソースは上記で参照しているチュートリアルから参照可能。

作成したプログラムは以下のとおり。

chap7-1.py

# -*- coding:utf-8 -*-
from pyspark.context import SparkContext
from pyspark.streaming import StreamingContext
from pyspark import StorageLevel

sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)
lines = ssc.socketTextStream("localhost", 9999,StorageLevel.MEMORY_AND_DISK_SER) 

words = lines.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

wordCounts.pprint()
ssc.start()             
ssc.awaitTermination() 

StreamingContextを生成

StreamingContextはストリーミング機能のメインエントリーポイントであり、StreamingContextを利用して入力ストリームデータの定義などを行う。

class pyspark.streaming.StreamingContext(sparkContext, batchDuration=None, jssc=None)

以下の例では1秒毎のバッチ処理を繰り返すStreamingContextを生成。

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)

DStreamの作成

このコンテキスト(StreamingContext)を使用して、DStreamを定義する。DStreamは時々刻々と生成されるRDDみたいなもので、データサーバから受信するストリームデータである。DStreamに含まれるそれぞれのレコードは1行のテキストである。
以下の例ではsocketTextStreamメソッドを使用し、localhostの9999番ポートに接続するように定義。

lines = ssc.socketTextStream("localhost", 9999)

SocketTextStreamは、TCPソース(hostname:port)からinputを生成する。データはTCPソケットを使用して受け取とられ、UTF-8でエンコードし、¥nをデリミタとした行単位でバイトで受け取る。

socketTextStream(hostname, port, storageLevel=StorageLevel(True, True, False, False, 2))

StorageLevelは次のURLを参照。　http://spark.apache.org/docs/latest/api/python/pyspark.html

word count処理

ストリームデータとして得られたテキストラインをスペース区切りで文字に分割する。

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

この例では、それぞれのテキストラインが複数の語(word)に分けられ、そしてwordのストリームがDStreamとなる。

次にwordをカウントする。
words DStreamを(word,1)ペアのDStreamにマップし、それからwordの頻度を得るためにreduce処理を行う。

pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

これで単語毎の出現回数を保持するDStreamが定義できた。

次に毎秒生成されるカウントをpprintメソッドで出力する。

wordCounts.pprint()

ここまででセットアップは完了したので、以下のコードを追加してストリーム処理を開始する。

ssc.start()             # 計算処理を開始
ssc.awaitTermination()  # stopメソッドによるcontextの終了や例外が発生するまで待つ

動作確認

作成したSparkStreamingのアプリを実行

$ spark-submit --master local chap7-1.py

別ターミナルにて、データ書き込み用に、NetCatを使って9999番ポートで待ち受けるようにし、テキストデータを書き込む。

$ nc -lk 9999
hello world

プログラムを実行したターミナルにて以下の出力があることを確認。

・・・
------------------------------------------
Time: 2015-12-22 08:28:02
-------------------------------------------
('world', 1)
('hello', 1)
・・・

HDFS環境でSpark Streamingの動作確認

HDFS上のディレクトリを監視し、新しいファイルが生成されたら、入力データとして取り込み、ワードをカウントするアプリを実行する。
※既存のファイルを更新しても入力データとして認識されないので注意。

変更点

先ほどのコードとの違いは以下の点。

• localの代わりにyarn-clientをmasterとして動作するように指定

sc = SparkContext("yarn-client", "Chap7-2")

• socketTextStreamメソッドの代わりにtextFileStreamメソッドを使用し、監視対象のhdfsディレクトリを引数で指定。textFileStreamはディレクトリに含まれるファイルを監視し、新しいファイルが生成されたらストリームデータとして取得する。

lines = ssc.textFileStream("hdfs:///user/y_tadayasu/data/") 

動作確認

アプリの実行

$ spark-submit --master yarn-client ./chap7-2.py

HDFS上にファイルを生成

$ hdfs dfs -put /opt/spark/README.md /user/y_tadayasu/data/

アプリを実行しているターミナルに以下の出力があることを確認。

-------------------------------------------
Time: 2015-12-23 04:15:50
-------------------------------------------
('', 67)
('guide,', 1)
('APIs', 1)
('name', 1)
('It', 2)
('package.', 1)
('particular', 3)
('tools', 1)
('must', 1)
('params', 1)
...

この結果だと空白もカウントされているので、filterを使用して空白は取り除くようにする。

words = lines.flatMap(lambda line: line.split(" ")).filter(lambda x:x)

-------------------------------------------
Time: 2015-12-23 04:25:30
-------------------------------------------
('guide,', 1)
('APIs', 1)
('optimized', 1)
('name', 1)
('It', 2)
('package.', 1)
('particular', 3)
('tools', 1)
('must', 1)
('params', 1)
...

コード全体は以下のとおり。

chap7-2.py

# -*- coding:utf-8 -*-
from pyspark.context import SparkContext
from pyspark.streaming import StreamingContext
from pyspark import StorageLevel
sc = SparkContext("yarn-client", "Chap7-2")
ssc = StreamingContext(sc, 10)
lines = ssc.textFileStream("hdfs:///user/y_tadayasu/data/") 
words = lines.flatMap(lambda line: line.split(" ")).filter(lambda x:x)
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)
wordCounts.pprint()
ssc.start()             
ssc.awaitTermination()