csv - Spark中CSV的多行标题

Question

我需要Spark将压缩的csv文件写入HDFS，但是我需要从几行版本信息开始。

文件内容示例

version=2
date=2020-01-31

id,name,age
1,Alice,21
2,Bob,23

三种实现方法的想法

首先写到hdfs://data/tmp/file1.csv.gz，然后使用hadoop fs -cat
将其流式传输到hdfs://data/real/file1.csv.gz

转换输出
将datafram转换为文本格式/ RDD[String]，然后将实际文件与
多余的标题行

将第一列名称更改为多行

所以对于方法3:
column1 ="version=2\ndate=2020-01-31\n\nid"
如果您知道更优雅的方法，请告诉我。

Answer 1

我尝试了所有方法。这是简化的代码:

方法1

方法1使用来自bash脚本等的Hadoop命令。

这可行，但是需要两次HDFS写入和清理。而且它在Scala Spark项目中不太适合。

(echo -e "version=2\ndate=2020-01-31\n\nid,name,age" | gzip  -vc ; hadoop fs -cat "$INPUT_DIR/*" ) | hadoop fs -put - "$OUTPUT_PATH"

这里发生的是它将

将多行标题回显到stdout

用管道将其导入gzip并输出到

将其他HDSF目录传递到stdout

管道到hadoop fs -put中，它将结合所有

方法2

代码稍微复杂一点，标题中没有不好的引号字符，但是标题有时出现在csv部分之后。

import org.apache.hadoop.io.compress.GzipCodec

val heading = """version=2
date=2020-01-31

id,name,age""".split("\n", -1).toSeq

val headingRdd: RDD[String] = sc.parallelize(heading)

val mediamathRdd: RDD[String] = df.rdd.map(row => row.mkString(","))

val combinedResult: RDD[String] = (headingRdd union mediamathRdd)

combinedResult.repartition(1).saveAsTextFile(path, classOf[GzipCodec])

方法3

最简单的方法，但输出略有下降

df.repartition(1)
.withColumnRenamed("id", "version=2\ndate=2020-01-31\n\nid")
.option("header", true)
.option("delimiter", ",")
.option("quoteMode", "NONE")
.option("quote", " ")
.option("codec", "gzip")
.csv(path)

结果看起来像这样，可能会或可能不会

 version=2
date=2020-01-31

id ,name,age
1,Alice,21
2,Bob,23

我也尝试过:
.option("quote", "\u0000")
它实际上打印出ascii宪章零，尽管这在我的HDFS查看器中没有显示，但这不是规范的一部分。

最佳方法

它们都不是看起来很简单的任务的完美选择。也许有一个小方法可以使方法2正常工作。