我正在使用spark-streaming阅读kafka流媒体消息 . 现在我想将Cassandra设置为输出 . 我在cassandra "test_table"中创建了一个表"key:text primary key"和"value:text"我已成功将数据映射到 JavaDStream<Tuple2<String,String>> data
,如下所示:
JavaSparkContext sc = new JavaSparkContext("local[4]", "SparkStream",conf);
JavaStreamingContext jssc = new JavaStreamingContext(sc, new Duration(3000));
JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, args[0], args[1], topicMap );
JavaDStream<Tuple2<String,String>> data = messages.map(new Function< Tuple2<String,String>, Tuple2<String,String> >()
{
public Tuple2<String,String> call(Tuple2<String, String> message)
{
return new Tuple2<String,String>( message._1(), message._2() );
}
}
);
然后我创建了一个List:
List<TestTable> list = new ArrayList<TestTable>();
其中TestTable是我的自定义类,具有与我的Cassandra表相同的结构,其成员为“key”和“value”:
class TestTable
{
String key;
String val;
public TestTable() {}
public TestTable(String k, String v)
{
key=k;
val=v;
}
public String getKey(){
return key;
}
public void setKey(String k){
key=k;
}
public String getVal(){
return val;
}
public void setVal(String v){
val=v;
}
public String toString(){
return "Key:"+key+",Val:"+val;
}
}
请建议一种如何将 JavaDStream<Tuple2<String,String>> data
中的数据添加到 List<TestTable> list
的方法 . 我这样做,以便我可以随后使用
JavaRDD<TestTable> rdd = sc.parallelize(list);
javaFunctions(rdd, TestTable.class).saveToCassandra("testkeyspace", "test_table");
将RDD数据保存到Cassandra中 .
我试过这种方式编码:
messages.foreachRDD(new Function<Tuple2<String,String>, String>()
{
public List<TestTable> call(Tuple2<String,String> message)
{
String k = message._1();
String v = message._2();
TestTable tbl = new TestTable(k,v);
list.put(tbl);
}
}
);
但似乎有些类型的错配发生 . 请帮忙 .
1 回答
假设此程序的目的是将流数据从kafka保存到Cassandra,则不必将
JavaDStream<Tuple2<String,String>>
数据转储到List<TestTable>
列表中 .DataStax的Spark-Cassandra连接器直接通过Spark Streaming extensions支持此功能 .
在
JavaDStream
上使用此类扩展应该足够了:而不是在中间列表上排放数据 .