如何运行多个Spark Cassandra查询
如何运行多个Spark Cassandra查询
提问于 2019-07-03 17:42:17
我需要在下面运行这样一个任务。不知何故，我漏掉了一点。我知道，我不能像这样使用javasparkcontext并传递javafunctions，因为有序列化问题。
我需要运行多个大小为cartesian.size()的cassandra查询。有什么建议吗？
JavaSparkContext jsc = new JavaSparkContext(conf);
JavaRDD<DateTime> dateTimeJavaRDD = jsc.parallelize(dateTimes); //List<DateTime>
JavaRDD<Integer> virtualPartitionJavaRDD = jsc.parallelize(virtualPartitions); //List<Integer>
JavaPairRDD<DateTime, Integer> cartesian = dateTimeJavaRDD.cartesian(virtualPartitionJavaRDD);
long c = cartesian.map(new Function<Tuple2<DateTime, Integer>, Long>() {
@Override
public Long call(Tuple2<DateTime, Integer> tuple2) throws Exception {
return javaFunctions(jsc).cassandraTable("keyspace", "table").where("p1 = ? and p2 = ?", tuple2._1(), tuple2._2()).count();
}
}).reduce((a,b) -> a + b);
System.out.println("TOTAL ROW COUNT IS: " + c);回答 1
Stack Overflow用户
回答已采纳
发布于 2019-07-03 18:54:19
正确的解决方案应该是在数据和Casasndra表之间执行连接。有一个joinWithCassandraTable function可以满足您的需求--您只需生成包含p1和p2值的Tuple2的RDD，然后调用joinWithCassandra表，就像这样(未经过测试，取自我的示例here)：
JavaRDD<Tuple2<Integer, Integer>> trdd = cartesian.map(new Function<Tuple2<DateTime, Integer>, Tuple2<Integer, Integer>>() {
@Override
public Tuple2<Integer, Integer> call(Tuple2<DateTime, Integer> tuple2) throws Exception {
return new Tuple2<Integer, Integer>(tuple2._1(), tuple2._2());
}
});
CassandraJavaPairRDD<Tuple2<Integer, Integer>, Tuple2<Integer, String>> joinedRDD =
trdd.joinWithCassandraTable("test", "jtest",
someColumns("p1", "p2"), someColumns("p1", "p2"),
mapRowToTuple(Integer.class, String.class), mapTupleToRow(Integer.class));
// perform counting here...页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/56867238
复制相关文章
相似问题
腾讯云开发者
