Spark Cassandra 连接器 keyBy 和洗牌

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>

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

所以如果我有 RK='profile1'，列名称='George' 和年龄='34'，结果 RDD 将是：

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

然后我需要将共享相同密钥的元素组合在一起以获得 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();
            }
        })

我的问题是：

在RDD上使用keyBy会造成shuffle，还是会把数据保存在本地？
有没有办法在不打乱的情况下按键对元素进行分组？我阅读了 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

Spark Cassandra 连接器 keyBy 和洗牌

Spark Cassandra Connector keyBy and shuffling

grouping

shuffle

connector

cassandra

apache-spark