有没有办法按行拆分 RDD？答案

【问题标题】：Is there a way to split an RDD by rows?有没有办法按行拆分 RDD？
【发布时间】：2019-11-09 10:41:46
【问题描述】：

我在 JavaRDD 中有一堆 20000 行的数据。现在我想保存几个大小完全相同的文件（比如每个文件 70 行）。

我用下面的代码试过了，但是因为它不能完全分割，一些数据集由 69、70 或 71 行组成。困难是我需要所有的大小都相同，除了最后一条记录（它可以更少）。

感谢您的帮助！！！提前谢谢各位！

myString.repartition(286).saveAsTextFile(outputPath);

【问题讨论】：

这并不是 Spark 的工作原理。我能问一下，为什么每个文件中的行数完全相同很重要吗？？
rdd.coalesce(((rdd.count()/70).toInt)).saveAsTextFile("directory_where_") 试试这个合并然后 saveAsTextFile
@rohitprakash 这对我不起作用。我的 16353 行仅分为两个文件，分别为 8242 和 8111 行。 trainDataFeatures.coalesce((int)(trainDataFeatures.count()/70)).saveAsTextFile(outputPathTrainFeatures);
@GlennieHellesSindholt 因为我想将我的数据放入 SequenceRecordReader 以拟合机器学习模型。

标签： java string apache-spark split rdd

【解决方案1】：

您可以使用 filterByRange 执行类似（伪代码）的操作：

for i = 0; i < javaRDD.size ; i+= 70
    val tempRDD = javaRDD.filterByRange(i,i+70).repartition(1)
    tempRDD.saveAsTextFile(outputPath + i.toString());

【讨论】：

我正在编写自己的答案，但这样更简洁:)
JavaRDD 没有 filterByRange。 spark.apache.org/docs/2.1.1/api/java/org/apache/spark/api/java/…我做错了什么？

【解决方案2】：

不幸的是，这是一个 Scala 答案，但它有效。

首先定义一个自定义分区器：

class IndexPartitioner[V](n_per_part: Int, rdd: org.apache.spark.rdd.RDD[_ <: Product2[Long, V]], do_cache: Boolean = true) extends org.apache.spark.Partitioner {

    val max = {
        if (do_cache) rdd.cache()
        rdd.map(_._1).max
    }

    override def numPartitions: Int = math.ceil(max.toDouble/n_per_part).toInt
    override def getPartition(key: Any): Int = key match {
        case k:Long => (k/n_per_part).toInt
        case _ => (key.hashCode/n_per_part).toInt
    }
}

创建一个随机字符串的 RDD 并对其进行索引：

val rdd = sc.parallelize(Array.tabulate(1000)(_ => scala.util.Random.alphanumeric.filter(_.isLetter).take(5).mkString))  
val rdd_idx = rdd.zipWithIndex.map(_.swap)

创建分区器并应用它：

val partitioner = new IndexPartitioner(70, rdd_idx)
val rdd_part = rdd_idx.partitionBy(partitioner).values

检查分区大小：

rdd_part
  .mapPartitionsWithIndex{case (i,rows) => Iterator((i,rows.size))}
  .toDF("partition_number","number_of_records")
  .show

/**
+----------------+-----------------+
|               0|               70|
|               1|               70|
|               2|               70|
|               3|               70|
|               4|               70|
|               5|               70|
|               6|               70|
|               7|               70|
|               8|               70|
|               9|               70|
|              10|               70|
|              11|               70|
|              12|               70|
|              13|               70|
|              14|               20|
+----------------+-----------------+
*/

每个分区一个文件：

import sqlContext.implicits._
rdd_part.toDF.write.format("com.databricks.spark.csv").save("/tmp/idx_part_test/")

（为“_SUCCESS”+1）

XXX$ ls /tmp/idx_part_test/ | wc -l
16

【讨论】：