高效 Spark Cassandra Java 加入

Question

我有两张桌子：

1. my_keyspace.name 与列：
   • 名称（字符串）- 分区键
   • 时间戳（日期） - 分区键的第二部分
   • id (int) - 分区键的第三部分
2. my_keyspace.data 与列：
   • 时间戳（日期）- 分区键
   • id (int) - 分区键的第二部分
   • 数据（字符串）

我正在尝试加入名称表中的时间戳和 ID。我通过获取与给定名称关联的所有时间戳和 id 并从数据表中检索这些条目的数据来做到这一点。

在 CQL 中执行此操作非常快。我预计 Spark Cassandra 的速度同样快，但它似乎是在进行全表扫描。这可能是由于不知道哪些字段是分区/主键。虽然我似乎无法找到一种方法来告诉它映射。

我怎样才能使这个连接尽可能高效？这是我的代码示例：

private static void notSoEfficientJoin() {
    SparkConf conf = new SparkConf().setAppName("Simple Application")
                                    .setMaster("local[*]")
                                    .set("spark.cassandra.connection.host", "localhost")
                                    .set("spark.driver.allowMultipleContexts", "true");
    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaPairRDD<DataKey, NameRow> nameIndexRDD = javaFunctions(sc).cassandraTable("my_keyspace", "name", mapRowTo(NameRow.class)).where("name = 'John'")
                                                                       .keyBy(new Function<NameRow, DataKey>() {
                                                                           @Override
                                                                           public DataKey call(NameRow v1) throws Exception {
                                                                               return new DataKey(v1.timestamp, v1.id);
                                                                           }
                                                                       });

    JavaPairRDD<DataKey, DataRow> dataRDD = javaFunctions(sc).cassandraTable("my_keyspace", "data", mapRowTo(DataRow.class))
                                                          .keyBy(new Function<DataRow, DataKey>() {
                                                              @Override
                                                              public DataKey call(DataRow v1) throws Exception {
                                                                  return new DataKey(v1.timestamp, v1.id);
                                                              }
                                                          });

    JavaRDD<String> cassandraRowsRDD = nameIndexRDD.join(dataRDD)
                                                       .map(new Function<Tuple2<DataKey, Tuple2<NameRow, DataRow>>, String>() {
                                                           @Override
                                                           public String call(Tuple2<DataKey, Tuple2<NameRow, DataRow>> v1) throws Exception {
                                                               NameRow nameRow = v1._2()._1();
                                                               DataRow dataRow = v1._2()._2();
                                                               return nameRow + " " + dataRow;
                                                           }
                                                       });

    List<String> collect = cassandraRowsRDD.collect();
}

Answer 1

更有效地执行此连接的方法是实际调用joinWithCassandraTable，这可以通过用另一个javaFunctions 调用包装结果来完成：

private static void moreEfficientJoin() {
    SparkConf conf = new SparkConf().setAppName("Simple Application")
                                    .setMaster("local[*]")
                                    .set("spark.cassandra.connection.host", "localhost")
                                    .set("spark.driver.allowMultipleContexts", "true");
    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD<DataKey> nameIndexRDD = sc.parallelize(javaFunctions(sc).cassandraTable("my_keyspace", "name", mapRowTo(DataKey.class))
                                                                    .where("name = 'John'")
                                                                    .collect());

    JavaRDD<Data> dataRDD = javaFunctions(nameIndexRDD).joinWithCassandraTable("my_keyspace", "data", allColumns, someColumns("timestamp", "id"), mapRowTo(Data.class), mapToRow(DataKey.class))
                                                       .map(new Function<Tuple2<DataKey, Data>, Data>() {
                                                           @Override
                                                           public Data call(Tuple2<DataKey, Data> v1) throws Exception {
                                                               return v1._2();
                                                           }
                                                       });
    List<Data> data = dataRDD.collect();
}

重要的是用javaFunctions 包裹JavaRDD。所以可以不调用collect和sc.parallelizenameIndexRDD