は、それが実現可能なスパーク労働者

Question

SparkStreamingContextがドライバーに作成され、それをシリアル化することはできませんから、カサンドラにSparkStreamingからデータを保存することです。したがって、私たちは労働者のこのオブジェクトにアクセスすることはできません。スパークの分散された性質を利用するために、私たちは労働者のデータ/ストリームを処理していました。したがって、OOMを提供する可能性のあるすべてのデータをドライバに収集するのではなく、作業者から直接データを保存することが理にかなっています。

私は私はいくつかのビジネス・ロジックを適用する必要があり、最終的にはカサンドラにデータを格納する必要があり、このようなシナリオを持っています。だから、どうやってやるの？

DStream.foreachRDD{ x => x.foreachPartition { some logic and finally Store the data to cassandra } } 

Answer 1

あなたは、カサンドラDBに直接JavaDStreamを保存することができますdatastaxの火花カサンドラ・コネクタ（https://github.com/datastax/spark-cassandra-connector）を使用することができます。

import static com.datastax.spark.connector.japi.CassandraStreamingJavaUtil.javaFunctions; 

Map<String, String> columnNameMappings; 
JavaDStream<MyTableData> myDStream; 
javaFunctions(myDStream).writerBuilder("mykeyspace", "my_table", 
      CassandraJavaUtil.mapToRow(MyTableData.class, columnNameMappings)).saveToCassandra(); 

Answer 2

val sparkConf = new SparkConf().setAppName("Test App") 
sparkConf.set("spark.cassandra.connection.host", "X.X.X.X") 
sparkConf.set("spark.cassandra.auth.username", "xxxxx")    
sparkConf.set("spark.cassandra.auth.password", "xxxxx") 

SparkConfあなたのカサンドラの接続構成を保持しているコンテキスト内のオブジェクトであるので、そこを設定します。

SparkContextはStreamingContextの一部ですが、あなたは本当にそれを心配する必要はありません。あなたがSparkContextまたは現在SparkConfにアクセスする必要があるなら、あなたはこの

StreamingContext ssc = new StreamingContext(sparkConf, Seconds(1)) 
ssc.sparkContext => Your Spark Context object 
ssc.sparkContext.getConf => Your SparkConf object 

はカサンドラ部分に保存する方法に来るように、それはこのように行うことができることを行うことができます。

// messages is the InputDStream in this example 
messages.foreachRDD(x => { 
    // Write business logic 
    x.saveToCassandra("keyspace_name", "table_name") 
}