从 Cassandra 将宽行分布式加载到 Spark答案

【问题标题】：Distributed loading of a wide row into Spark from Cassandra从 Cassandra 将宽行分布式加载到 Spark
【发布时间】：2014-11-25 01:08:39
【问题描述】：

假设我们有一个 RF = N 的 Cassandra 集群和一个包含宽行的表。

我们的表可能有一个类似这样的索引：pk / ck1 / ck2 / ....

如果我们从表中的一行创建一个RDD如下：

val wide_row = sc.cassandraTable(KS, TABLE).select("c1", "c2").where("pk = ?", PK)

我注意到一个 Spark 节点拥有 100% 的数据，而其他节点则没有。我认为这是因为 spark-cassandra-connector 无法将查询令牌范围分解为更小的子范围，因为它实际上不是一个范围——它只是 PK 的哈希。

此时，我们可以简单地调用redistribute(N)，在处理之前将数据分散到 Spark 集群中，但这会导致将数据通过网络移动到 Cassandra 本地已经拥有数据的节点（记住 RF = N )

我们真正想要的是让每个 Spark 节点从 Cassandra 本地加载行的子集（切片）。

想到的一种方法是在 pk = PK 时生成一个包含第一个集群键 (ck1) 的不同值列表的 RDD。然后，我们可以使用mapPartitions() 根据 ck1 的每个值加载宽行的切片。

假设我们已经有了 ck1 的列表值，我们可以这样写：

val ck1_list = ....  // RDD

ck1_list.repartition(ck1_list.count().toInt) // create a partition for each value of ck1

val wide_row = ck1_list.mapPartitions(f)

在分区迭代器 f() 中，我们想调用另一个函数 g(pk, ck1)，它从 Cassandra 加载行切片，用于分区键 pk 和集群键 ck1。然后我们可以将flatMap 应用到ck1_list，从而创建一个完全分布式的宽行RDD，而无需任何洗牌。

那么问题来了：

是否可以从 Spark 任务中进行 CQL 调用？应该使用什么驱动程序？是否只能设置一次以重复用于后续任务？

任何帮助将不胜感激，谢谢。

【问题讨论】：

标签： scala cassandra apache-spark

【解决方案1】：

为了以后参考，我将解释我是如何解决这个问题的。

我实际上使用了一种与上述方法略有不同的方法，它不涉及从 Spark 任务中调用 Cassandra。

我从 ck_list 开始，这是 pk = PK 时第一个集群键的不同值的列表。代码在这里没有显示，但我实际上是使用 CQL 直接从 Spark 驱动程序中的 Cassandra 下载的。

然后我将 ck_list 转换为 RDDS 列表。接下来我们将 RDD（每个代表一个 Cassandra 行切片）组合成一个统一的 RDD (wide_row)。

CassandraRDD 的强制转换是必要的，因为 union 返回类型 org.apache.spark.rdd.RDD

运行作业后，我能够验证wide_row 有 x 个分区，其中 x 是 ck_list 的大小。一个有用的副作用是wide_row 被第一个集群键分区，这也是我想要减少的键。因此，避免了更多的洗牌。

我不知道这是否是实现我想要的最佳方式，但它确实有效。

val ck_list // list first cluster key values where pk = PK

val wide_row = ck_list.map( ck =>
  sc.cassandraTable(KS, TBL)
    .select("c1", "c2").where("pk = ? and ck1 = ?", PK, ck)
    .asInstanceOf[org.apache.spark.rdd.RDD] 
).reduce( (x, y) => x.union(y) )

【讨论】：