Spark SQL 如何读取压缩后的 csv 文件？

Question

我曾与API试图spark.read.csv读取与扩展压缩csv文件bz或gzip。有效。但是在源代码中我没有找到任何可以声明codec类型的选项参数。

即使在这个链接中，也只有codec写作方面的设置。谁能告诉我或提供显示 spark 2.x 版本如何处理压缩 csv 文件的源代码的路径。

Answer 1

所有与文本相关的数据源，包括CSVDataSource，都使用 Hadoop File API 来处理文件（Spark Core 的 RDD 中也有）。

您可以在readFile中找到导致HadoopFileLinesReader的相关行，其中包含以下行：

val fileSplit = new FileSplit(
  new Path(new URI(file.filePath)),
  file.start,
  file.length,
  // TODO: Implement Locality
  Array.empty)

它使用 Hadoop 的org.apache.hadoop.fs.Path来处理底层文件的压缩。

经过快速谷歌搜索后，我找到了处理压缩的 Hadoop 属性，即mapreduce.output.fileoutputformat.compress.

这导致我使用 Spark SQL 的CompressionCodecs进行以下压缩配置：

"none" -> null,
"uncompressed" -> null,
"bzip2" -> classOf[BZip2Codec].getName,
"deflate" -> classOf[DeflateCodec].getName,
"gzip" -> classOf[GzipCodec].getName,
"lz4" -> classOf[Lz4Codec].getName,
"snappy" -> classOf[SnappyCodec].getName)

在下面的代码中，您可以找到使用“our”选项的setCodecConfiguration 。

  def setCodecConfiguration(conf: Configuration, codec: String): Unit = {
    if (codec != null) {
      conf.set("mapreduce.output.fileoutputformat.compress", "true")
      conf.set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString)
      conf.set("mapreduce.output.fileoutputformat.compress.codec", codec)
      conf.set("mapreduce.map.output.compress", "true")
      conf.set("mapreduce.map.output.compress.codec", codec)
    } else {
      // This infers the option `compression` is set to `uncompressed` or `none`.
      conf.set("mapreduce.output.fileoutputformat.compress", "false")
      conf.set("mapreduce.map.output.compress", "false")
    }
  }

另一个方法getCodecClassName用于解析JSON、CSV和文本compression格式的选项。