Sparkストリーミングの概要

Sparkストリーミングを使用すると、Akka、Kafka、Twitterなどのソースからの生データ・ストリームを消費できます。その後、これらのデータはSparkアプリケーションによって分析され、データベースに格納できます。この例では、Scalaを使用します。

1. ストリーミング・コンテキスト・オブジェクトをインポートします。

   import org.apache.spark.streaming._

2. 新しいStreamingContextオブジェクトを既存のSparkConf構成オブジェクトに基づいて作成します。この際、バッチの持続期間の値を渡すことによって、ストリーミング・データをバッチに分割する間隔を指定します。

   val sparkConf = ....
   val ssc = new StreamingContext(sc, Seconds(1)) // Uses the context automatically created by the spark shell

   Sparkでは、バッチの持続期間をミリ秒、秒、分の各単位で指定できます。
3. StreamingContext、DStream、RDDの各オブジェクトに関するデータベース特有の関数をインポートします。

   import com.datastax.spark.connector.streaming._

4. データ・ストリームを提供するサービスのIPとポートに接続するDStreamオブジェクトを作成します。

   val lines = ssc.socketTextStream(server IP address, server port number)

5. 各バッチの単語数をカウントし、データをテーブルに保存します。

   val words = lines.flatMap(_.split(" "))
   val pairs = words.map(word => (word, 1))
   val wordCounts = pairs.reduceByKey(_ + _)    
       .saveToCassandra("streaming_test", "words_table", SomeColumns("word", "count"))

6. 計算を開始します。

   ssc.start()
   ssc.awaitTermination()

CREATE KEYSPACE IF NOT EXISTS streaming_test 
WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }; 

CREATE TABLE IF NOT EXISTS streaming_test.words_table 
(word TEXT PRIMARY KEY, count COUNTER);

nc -lk 9999
one two two three three three four four four four someword

dse spark

import org.apache.spark.streaming._
import com.datastax.spark.connector.streaming._

val ssc =  new StreamingContext(sc, Seconds(1)) 
val lines = ssc.socketTextStream( "localhost", 9999)
val words = lines.flatMap(_.split( " "))
val pairs = words.map(word => (word, 1))

val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.saveToCassandra( "streaming_test",  "words_table", SomeColumns( "word",  "count"))
wordCounts.print()
ssc.start()
ssc.awaitTermination()
exit()

cqlsh -k streaming_test

select * from words_table;

word    | count
---------+-------
   three |     3
     one |     1
     two |     2
    four |     4
someword |     1