scala - Apache Spark 的 RDD 根据特定大小拆分

Question

我正在尝试从文本文件中读取字符串，但我想根据特定大小限制每一行。例如;

这是我代表文件。

aaaaa\nbbb\nccccc

当试图通过 sc.textFile 读取这个文件时，RDD 会出现这个。

scala> val rdd = sc.textFile("textFile")
scala> rdd.collect
res1: Array[String] = Array(aaaaa, bbb, ccccc)

但是我想限制这个RDD的大小。例如，如果限制为 3，那么我应该像这样。

Array[String] = Array(aaa, aab, bbc, ccc, c)

做到这一点的最佳性能方式是什么？

Answer 1

不是一个特别有效的解决方案(也不可怕)，但您可以执行以下操作:

val pairs = rdd
  .flatMap(x => x)  // Flatten
  .zipWithIndex  // Add indices
  .keyBy(_._2 / 3)  // Key by index / n

// We'll use a range partitioner to minimize the shuffle 
val partitioner = new RangePartitioner(pairs.partitions.size, pairs)

pairs
  .groupByKey(partitioner)  // group
  // Sort, drop index, concat
  .mapValues(_.toSeq.sortBy(_._2).map(_._1).mkString("")) 
  .sortByKey()
  .values

可以通过传递显式填充分区所需的数据来避免混洗，但需要一些努力来编写代码。查看我对 Partition RDD into tuples of length n 的回答.

如果您可以接受分区边界上一些未对齐的记录，那么简单 mapPartitions with grouped 应该以低得多的成本来解决这个问题:

rdd.mapPartitions(_.flatMap(x => x).grouped(3).map(_.mkString("")))

也可以使用滑动 RDD:

rdd.flatMap(x => x).sliding(3, 3).map(_.mkString(""))