在Spark中保存时，在文本文件之前添加标题

Question

我有一些火花代码来处理csv文件。它对其进行一些转换。我现在想将此RDD保存为csv文件并添加一个标头。此RDD的每一行都已正确格式化。

我不知道怎么做。我想用头字符串和我的RDD进行联合，但头字符串不是RDD，因此无法工作。

Answer 1

def addHeaderToRdd(sparkCtx: SparkContext, lines: RDD[String], header: String): RDD[String] = {

    val headerRDD = sparkCtx.parallelize(List((-1L, header)))     // We index the header with -1, so that the sort will put it on top.

    val pairRDD = lines.zipWithIndex()

    val pairRDD2 = pairRDD.map(t => (t._2, t._1))

    val allRDD = pairRDD2.union(headerRDD)

    val allSortedRDD = allRDD.sortByKey()

    return allSortedRDD.values
}

Answer 2

关于在没有联合的情况下编写它的一些帮助（在合并时提供了标题）

val fileHeader ="This is header"
val fileHeaderStream: InputStream = new  ByteArrayInputStream(fileHeader.getBytes(StandardCharsets.UTF_8));
val output = IOUtils.copyBytes(fileHeaderStream,out,conf,false)

现在循环遍历文件部分以使用

val in: DataInputStream = ...<data input stream from file >
 IOUtils.copyBytes(in, output, conf, false)

这对我来说确保了标题总是作为第一行，即使您使用coalasec/重新分区进行高效编写

Answer 3

您可以从标题行中制作一个RDD，然后Union它，是的：

val rdd: RDD[String] = ...
val header: RDD[String] = sc.parallelize(Array("my,header,row"))
header.union(rdd).saveAsTextFile(...)

然后，您将得到一组合并的文件。

问题是，我认为不能保证头是第一个分区，因此最终会出现在第00000部分，并位于文件的顶部。实际上，我很肯定它会的。

更可靠的方法是使用Hadoop命令，如hdfs来合并part-xxxxx文件，作为命令的一部分，只需插入文件的标题行即可。