cassandra - Spark Cassandra 连接器 keyBy 和 shuffle

Question

我正在尝试通过尽可能避免洗牌来优化我的 Spark 工作。

我正在使用 cassandraTable 创建 RDD。

列族的列名是动态的，因此定义如下:

CREATE TABLE "Profile" (
  key text,
  column1 text,
  value blob,
  PRIMARY KEY (key, column1)
) WITH COMPACT STORAGE AND
  bloom_filter_fp_chance=0.010000 AND
  caching='ALL' AND
  ...

此定义会产生以下格式的 CassandraRow RDD 元素:

CassandraRow <key, column1, value>

• 键 - RowKey
• column1 - column1 的值是动态列的名称
• value - 动态列的值

因此，如果我有 RK='profile1'，其中列 name='George' 和age='34'，则生成的 RDD 将是:

CassandraRow<key=profile1, column1=name, value=George>
CassandraRow<key=profile1, column1=age, value=34>

然后我需要将共享相同 key 的元素分组在一起以获得 PairRdd:

PairRdd<String, Iterable<CassandraRow>>

重要的是，我需要分组的所有元素都在同一个 Cassandra 节点中(共享相同的行键)，因此我希望连接器保持数据的局部性。

问题在于使用 groupBy 或 groupByKey 会导致随机播放。我宁愿在本地对它们进行分组，因为所有数据都在同一个节点上:

JavaPairRDD<String, Iterable<CassandraRow>> rdd = javaFunctions(context)
        .cassandraTable(ks, "Profile")
        .groupBy(new Function<ColumnFamilyModel, String>() {
            @Override
            public String call(ColumnFamilyModel arg0) throws Exception {
                return arg0.getKey();
            }
        })

我的问题是:

1. 在 RDD 上使用 keyBy 会导致改组，还是会将数据保留在本地？
2. 有没有一种方法可以按键对元素进行分组而不需要打乱顺序？我读过有关mapPartitions的内容，但不太明白它的用法。

谢谢

谢

Answer 1

我认为您正在寻找 spanByKey，这是一个 cassandra 连接器特定的操作，它利用 cassandra 提供的排序来允许对元素进行分组，而不会导致随机播放阶段。

就您而言，它应该如下所示:

sc.cassandraTable("keyspace", "Profile")
  .keyBy(row => (row.getString("key")))
  .spanByKey

在文档中阅读更多内容:
https://github.com/datastax/spark-cassandra-connector/blob/master/doc/3_selection.md#grouping-rows-by-partition-key