Java Spark Streaming with Cassandra

Question

私はCassandraでJavaスパークストリーミングをしようとしています。私はScalaでも同じことをしましたが、私はJavaでどのように進めるべきかわかりません。ウェブは、私にJavaのスパークストリーミングとCassandraの例を教えてくれませんでした。

は、いくつかのいずれかは、Javaで以下のScalaのコードを持っているかを見せてもらえ：

import org.apache.spark.streaming.dstream.ConstantInputDStream 

val ssc = new StreamingContext(conf, Seconds(10)) 

val cassandraRDD = ssc.cassandraTable("mykeyspace", "users").select("fname", "lname").where("lname = ?", "yu") 

val dstream = new ConstantInputDStream(ssc, cassandraRDD) 

dstream.foreachRDD{ rdd => 
    // any action will trigger the underlying cassandra query, using collect to have a simple output 
    println(rdd.collect.mkString("\n")) 
} 
ssc.start() 
ssc.awaitTermination() 

すべてのヘルプは高く評価されます。ありがとう

Answer 1

あなたのforeachRDD変換では、あなたのデータをcassandraテーブル形式で変換できます。

JavaRDD<TestBean> cassandraRDD = testRDD 
       .flatMap(new FlatMapFunction<Tuple2<String, List<Map<String, Object>>>, TestBean>() { 

        private static final long serialVersionUID = 1L; 

        @Override 
        public Iterable<TestBean> call(Tuple2<String, List<Map<String, Object>>> tuple) throws Exception { 

         return rawData; 
        } 
       }); 

      javaFunctions(jsonRDD).writerBuilder(CASSANDRA_KEYSPACE,CASSANDRA_TABLE, mapToRow(TestBean.class)).saveToCassandra();