Spark：将 RDD 元素拆分为块

Question

我在 Scala 中编写了一个相对简单的 Spark 作业，它从 S3 读取一些数据，执行一些转换和聚合，最后将结果存储到存储库中。

在最后阶段，我有一个域模型的 RDD，我想将它们分组为元素块，以便我可以在我的存储库中进行一些大规模插入。

我使用RDDFunctions.sliding 方法来实现这一点，它几乎可以正常工作。这是我的代码的简化版本：

val processedElements: RDD[DomainModel] = _
RDDFunctions.fromRDD(processedElements)
    .sliding(500, 500)
    .foreach { elementsChunk =>
        Await.ready(repository.bulkInsert(elementsChunk), 1.minute)
    }

问题是，例如，如果我有 1020 个元素，那么我的存储库中只有 1000 个元素。如果窗口大小大于剩余元素的数量，看起来滑动会忽略任何其他元素。

有没有办法解决这个问题？如果没有，有没有其他方法可以在不使用RDDFunctions.sliding 的情况下实现相同的行为？

Answer 1

您不能只使用foreachPartition 和手动批处理管理吗？

fromRDD.foreachPartition(items: Iterator[DomainModel] => {
   val batch = new ArrayBuffer[DomainModel](BATCH_SIZE)
   while (items.hasNext) {
     if (batch.size >= BATCH_SIZE) {
       bulkInsert(batch)
       batch.clear()
     }
     batch += items.next
   }
   if (!batch.isEmpty) {
      bulkInsert(batch)
   }
})

Answer 2

根据RDDFunctions doc，如果窗口大小超过剩余项目的数量，Spark 的sliding（与 Scala 不同）会生成一个空 RDD，这是对的。 Spark 也没有 Scala 的grouped 等价物。

如果您知道要创建多少组，一个可能适用的解决方法是使用 modulo 过滤器拆分 RDD。这是一个将 RDD 分成 5 组的简单示例：

val rdd = sc.parallelize(Seq(
  (0, "text0"), (1, "text1"), (2, "text2"), (3, "text2"), (4, "text2"), (5, "text5"),
  (6, "text6"), (7, "text7"), (8, "text8"), (9, "text9"), (10, "text10"), (11, "text11")
))

def g(n:Int)(x: Int): Boolean = { x % 5 == n }

val rddList = (0 to 4).map( n => rdd.filter(x => g(n)(x._1)) )

(0 to 4).foreach(n => rddList(n).collect.foreach(println))

(0,text0)
(5,text5)
(10,text10)

(1,text1)
(6,text6)
(11,text11)

(2,text2)
(7,text7)

(3,text2)
(8,text8)

(4,text2)
(9,text9)