使用 Cassandra 进行 Java Spark 流式处理

Question

我正在尝试使用 Cassandra 进行 java spark 流式传输。我对 Scala 做了同样的事情，但我不知道如何在 Java 中进行。网络没有给我任何关于 java spark 流和 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();