Многострочный заголовок для CSV в Spark - PullRequest
Новая
       6

Многострочный заголовок для CSV в Spark

0 голосов
/ 31 января 2020

Мне нужно, чтобы Spark записал сжатый CSV-файл в HDFS, но мне нужно, чтобы он начинался с нескольких строк информации о версии.

Пример содержимого файла 

version=2
date=2020-01-31

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

Три идеи способов сделать это

  1. Сначала запишите в hdfs: //data/tmp/file1.csv.gz, затем используйте has 1025 * fs -cat для потоковой передачи в hdfs: //data/real/file1.csv.gz
  2. Преобразование выходного фрейма данных в текстовый формат / RDD[String] и объединение реального файла с дополнительными строками заголовка
  3. Изменение имени первого столбца на несколько строк

Так что для подхода 3:

column1 ="version=2\ndate=2020-01-31\n\nid"

Дайте мне знать, если вы знаете более элегантный способ сделать это.

1 Ответ

0 голосов
/ 03 февраля 2020

Я перепробовал все подходы. Вот упрощенный код:

Подход 1

Подход 1 использует команды Had oop из сценария bash и др. c.

Это работает, но требует двойной записи 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"

Здесь происходит то, что он

  1. выводит многострочный заголовок в стандартный вывод
  2. передать это в gzip и на стандартный вывод
  3. передать другое HDSF-каталог на стандартный вывод
  4. передать в 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 viewer это не было частью spe c.

Наилучший подход

Ни один из них не идеально подходит для того, что кажется очень простой задачей. Возможно, есть небольшое исправление, чтобы подход 2 работал идеально.

...