如何有效地使用 spark 读取具有分区热点的 cassandra 数据？答案

【问题标题】：How to effectively use spark to read cassandra data that has partition hotspots?如何有效地使用 spark 读取具有分区热点的 cassandra 数据？
【发布时间】：2017-06-10 05:51:51
【问题描述】：

据我所知，从 cassandra 读取数据时，spark 每个 cassandra 分区最多使用一个任务。不幸的是，我在 cassandra 中有一些非常不平衡的分区（糟糕的初始表设计）。我需要将该数据读入一个新表，该表将更好地处理热点，但任何尝试使用正常的火花途径都不会有效；我剩下几个任务（10 多个）永远运行，处理那些巨大的分区键。

为了让您了解规模，这是在一个大小约为 1.5TB 的表上工作，分布在 5 台服务器上，复制因子为 3； ~ 每个节点 500GB。

欢迎其他想法，但仅转储到 CSV 可能不是一个现实的选择。

到目前为止，物化视图创建也是不行的；它花费的时间太长了，至少在 3.0.8 上，在创建过程中几乎没有监控。

【问题讨论】：

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

【解决方案1】：

这是一个无法真正自动解决的难题但是如果你知道你的数据在你真正巨大的文件中是如何分布的，我可以给你一个选择。

与其使用单个 RDD/DataFrame 来表示您的表，不如将其拆分为多个联合调用。

基本上你想这样做

鉴于我们最大的分区是这样设置的

Key1 -> C1, C2, C3, ..., C5000000

我们知道 C 的分布一般是这样的

Min C = 0
Max C = 5000000
Average C = 250000

我们可以猜测，通过每 100K C 值执行一次范围下推，我们可以很好地分割这些大分区。

val interval = 100000
val maxValue = 500000
sc.union(
 (0 until maxValue by interval).map{ lowerBound => 
   sc.cassandraTable("ks", "tab")
     .where(s"c > $lowerBound AND c < ${lowerBound + interval}")
  }
)

我们最终会得到更多更小的分区（可能还有很多空的分区），但这应该让我们成功地削减那些巨大的分区。但是，只有在您能够弄清楚分区中值的分布情况下才能做到这一点。

注意:: union-ing 数据框也可以实现同样的事情

【讨论】：

据我知道，可以肯定的是，每个分区的最大范围，这可能是一个选项。我也可以做一个火花 DF.groupBy('partitionKey).count.describe，它应该给我分布。分区键实际上是第一个集群键的前缀，这使得这种方法更加容易。我仍然认为这是 spark-cassandra-connector 中的一个缺陷，找到解决它的标准方法会很有趣。
好吧，问题是不知道 c 的分布，就无法知道在哪里建立截止点。直到 Cassandra 中有近似的列统计信息，才能自动完成很多事情。但关键是你在阅读整个内容之前先进行切片和联合。