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

Question

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

即使在这个 link ，写入端只有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)

在下面的代码中，您可以找到 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 ，和text格式。