如何运行多个 Spark Cassandra 查询答案

【问题标题】：How To Run Multiple Spark Cassandra Query如何运行多个 Spark Cassandra 查询
【发布时间】：2019-11-13 23:22:02
【问题描述】：

我需要在下面运行这样的任务。不知何故，我错过了一点。我知道，我不能像这样使用 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);

【问题讨论】：

标签： java apache-spark spark-cassandra-connector

【解决方案1】：

正确的解决方案应该是在您的数据和 Casasndra 表之间执行连接。有joinWithCassandraTable function 正在做您需要的事情 - 您只需生成包含p1 和p2 的值的Tuple2 的RDD，然后调用joinWithCassandra 表，类似这样（未经测试，从我的示例中采用@ 987654322@):

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...

【讨论】：