如何使用wholeTextFiles读取Spark中的gz文件

Question

我有一个文件夹，其中包含许多小的.gz文件(压缩的csv文本文件)。我需要在我的Spark作业中读取它们，但问题是我需要根据文件名中的信息进行一些处理。因此，我没有使用：

JavaRDD<<String>String> input = sc.textFile(...)

因为据我所知，我不能以这种方式访问文件名。相反，我使用：

JavaPairRDD<<String>String,String> files_and_content = sc.wholeTextFiles(...);

因为这样我就得到了一对文件名和内容。然而，这样看来，输入阅读器无法从gz文件中读取文本，而是读取二进制乱码。

因此，我想知道是否可以将其设置为以某种方式读取文本，或者使用sc.textFile(...)访问文件名

Answer 1

你不能用wholeTextFiles读取gzipped文件，因为它使用的CombineFileInputFormat不能读取gzipped文件，因为they are not splittable (源代码证明)：

  override def createRecordReader(
      split: InputSplit,
      context: TaskAttemptContext): RecordReader[String, String] = {

    new CombineFileRecordReader[String, String](
      split.asInstanceOf[CombineFileSplit],
      context,
      classOf[WholeTextFileRecordReader])
  }

您可以结合使用newAPIHadoopFile和wholefileinputformat (不是内置在hadoop中，而是在整个互联网上)，以使其正常工作。

更新1:我认为WholeFileInputFormat不会工作，因为它只获取文件的字节，这意味着你可能必须编写自己的类，可能是扩展WholeFileInputFormat来确保它解压缩字节。

另一种选择是使用GZipInputStream自己解压缩字节

更新2:如果您有权访问目录名称，如下面的OP注释所示，您可以获得如下所示的所有文件。

Path path = new Path("");
FileSystem fileSystem = path.getFileSystem(new Configuration()); //just uses the default one
FileStatus []  fileStatuses = fileSystem.listStatus(path);
ArrayList<Path> paths = new ArrayList<>();
for (FileStatus fileStatus : fileStatuses) paths.add(fileStatus.getPath());

Answer 2

我在使用spark连接到S3时也遇到了同样的问题。

我的文件是一个没有扩展名的gzip csv。

JavaPairRDD<String, String> fileNameContentsRDD = javaSparkContext.wholeTextFiles(logFile);

此方法返回损坏的值

我用下面的代码解决了这个问题：

JavaPairRDD<String, String> fileNameContentsRDD = javaSparkContext.wholeTextFiles(logFile+".gz");

通过向S3地址添加.gz，spark自动选择文件并像读取gz文件一样读取它。(这似乎是一个错误的方法，但解决了我的问题。