Spark - 读取没有文件扩展名的压缩文件答案

【问题标题】：Spark - Read compressed files without file extensionSpark - 读取没有文件扩展名的压缩文件
【发布时间】：2016-10-27 13:31:36
【问题描述】：

我有一个 S3 存储桶，里面装满了没有文件扩展名的 Gz 文件。例如s3://mybucket/1234502827-34231

sc.textFile 使用该文件扩展名来选择解码器。我发现了很多关于处理自定义文件扩展名的博客文章，但没有关于缺少文件扩展名的内容。

我认为解决方案可能是sc.binaryFiles 并手动解压缩文件。

另一种可能性是弄清楚sc.textFile 是如何找到文件格式的。我不清楚这些classOf[] 调用的作用。

  def textFile(
      path: String,
      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
    assertNotStopped()
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
      minPartitions).map(pair => pair._2.toString).setName(path)
  }

【问题讨论】：

sc.textFile 无法确定格式。由TextInputFormat 完成，仅使用扩展名。
或者重命名s3中的所有文件，加上.gz。我查看了源代码，它在此处实现：hadoop.apache.org/docs/stable/api/src-html/org/apache/hadoop/io/… 它确实使用了文件扩展名。该规范建议您可以只查看前几个字节 zlib.org/rfc-gzip.html#file-format ，但这表明您可能会得到误报并且必须考虑字节序 stackoverflow.com/questions/6059302/… 所以毫无疑问，使用 .gz 是一种更快、更可靠的约定跨度>
@user6022341 TextInputFormat 没有这样做，是这个类中的getCodec(Path file) 方法org.apache.hadoop.io.compress.CompressionCodecFactory

标签： apache-spark

【解决方案1】：

您可以尝试将以下 ZIP 文件解决方案与 gzipFileInputFormat 库结合起来吗？

这里 - How to open/stream .zip files through Spark? 你可以看看如何使用 ZIP：

rdd1  = sc.newAPIHadoopFile("/Users/myname/data/compressed/target_file.ZIP", ZipFileInputFormat.class, Text.class, Text.class, new Job().getConfiguration());

gzipFileInputFormat：

https://github.com/bsankaran/internet_routing/blob/master/hadoop-tr/src/main/java/edu/usc/csci551/tools/GZipFileInputFormat.java

关于 newAPIHadoopFile() 的一些细节可以在这里找到： http://spark.apache.org/docs/latest/api/python/pyspark.html

【讨论】：

谢谢，我给了这几个小时但没能成功。

【解决方案2】：

我发现了几个几乎符合我需要的例子。这是我用来解析 GZ 压缩文件的最终代码。

import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream
import org.apache.spark.input.PortableDataStream
import scala.util.Try
import java.nio.charset._

def extractBSM(ps: PortableDataStream, n: Int = 1024) = Try {
  val gz = new GzipCompressorInputStream(ps.open)
  Stream.continually {
    // Read n bytes
    val buffer = Array.fill[Byte](n)(-1)
    val i = gz.read(buffer, 0, n)
    (i, buffer.take(i))
  }
  // Take as long as we've read something
  .takeWhile(_._1 > 0)
  .map(_._2)
  .flatten
  .toArray
}
def decode(charset: Charset = StandardCharsets.UTF_8)(bytes: Array[Byte]) = new String(bytes, StandardCharsets.UTF_8)
val inputFile = "s3://my-bucket/157c96bd-fb21-4cc7-b340-0bd4b8e2b614"
val rdd = sc.binaryFiles(inputFile).flatMapValues(x => extractBSM(x).toOption).map( x => decode()(x._2) )
val rdd2 = rdd.flatMap { x => x.split("\n") }
rdd2.take(10).foreach(println)

【讨论】：

这适用于 GZ，但我们确实需要检查魔术字节并应用正确的压缩算法
您应该注明来源。此外，这仅适用于不可拆分的格式（如 gz），因此具有识别功能的通用解决方案没有多大意义。
这个答案有一个正确性错误！即使我们没有到达流的末尾，read() 调用返回 0 字节在技术上也是有效的。我认为这应该读到你打到流结束（当-1从read()返回时）。更好的是，像 Guava 的 ByteStreams.toByteArray() 这样的辅助函数。

【解决方案3】：

您可以创建自己的自定义编解码器来解码您的文件。您可以从扩展 GzipCodec 开始并覆盖 getDefaultExtension 方法，在该方法中返回空字符串作为扩展名。

编辑：由于 CompressionCodecFactory 的实现方式，该灵魂不会在所有情况下都有效。例如：默认情况下会加载 .lz4 的编解码器。这意味着如果您要加载的文件的名称以 4 结尾，则将选择该编解码器而不是自定义（无扩展名）。由于该编解码器与扩展名不匹配，因此稍后会被丢弃，并且不会使用任何编解码器。

Java：

package com.customcodec;

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

public class GzipCodecNoExtension extends GzipCodec {

    @Override
    public String getDefaultExtension() {
        return "";
    }
}

在 Spark 应用中，您只需注册您的编解码器：

    SparkConf conf = new SparkConf()
            .set("spark.hadoop.io.compression.codecs", "com.customcodec.GzipCodecNoExtension");

【讨论】：

【解决方案4】：

您可以使用map函数读取二进制文件并进行解压。

JavaRDD<Tuple2<String, PortableDataStream>> rawData = spark.sparkContext().binaryFiles(readLocation, 1).toJavaRDD();

JavaRDD<String> decompressedData = rawData.map((Function<Tuple2<String, PortableDataStream>, String>) stringPortableDataStreamTuple2 -> {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    GZIPInputStream s = new GZIPInputStream(new ByteArrayInputStream(stringPortableDataStreamTuple2._2.toArray()));
    IOUtils.copy(s, out);

    return new String(out.toByteArray());
});

如果是 JSON 内容，您可以使用

将其读入数据集

Dataset co = spark.read().json(decompressedData);

【讨论】：