如何使用 pyspark 跳过损坏的 gzip？答案

【问题标题】：how to skip corrupted gzips with pyspark?如何使用 pyspark 跳过损坏的 gzip？
【发布时间】：2016-04-25 10:13:44
【问题描述】：

我需要从 hdfs 中读取大量 gzip，如下所示： sc.textFile('*.gz') 虽然其中一些 gzip 已损坏，但会引发

java.io.IOException: gzip 流 CRC 失败

停止整个进程的运行。

我阅读了here 的辩论，其中有人有相同的需求，但没有得到明确的解决方案。由于在 spark 中实现此功能是不合适的（根据链接），有没有办法只是粗暴地跳过损坏的文件？似乎对 scala 用户有提示，不知道如何在 python 中处理它。

或者我只能先检测损坏的文件，然后删除它们？

如果我有大量 gzip，运行一天后，发现最后一个已损坏怎么办。一整天都浪费了。并且 gzip 损坏很常见。

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

您可以手动列出所有文件，然后读取地图 UDF 中的文件。然后，UDF 可以使用 try/except 块来处理损坏的文件。

代码看起来像

import gzip
from pyspark.sql import Row

def readGzips(fileLoc):
    try:
        ...
        code to read file
        ...
        return record
    except:
        return Row(failed=fileLoc)

from os import listdir
from os.path import isfile, join
fileList = [f for f in listdir(mypath) if isfile(join(mypath, f))]

pFileList = sc.parallelize(fileList)
dataRdd = pFileList.map(readGzips).filter((lambda x: 'failed' not in x.asDict()))

【讨论】：

那么“读取文件的代码”部分是使用python的gzip模块解压文件吗？并且输入是路径字符串列表，我的文件在 hdfs 上，如何使用原始 python 读取它们？
您是在问如何从 hdfs 中读取文件吗？或者如何阅读 gzip？ stackoverflow.com/questions/10566558/… 描述了如何使用 gzip。并且 hdfs 应该可以通过“hdfs://path/to/file.gz”访问
我的意思是在 readGzips() 中，你只能使用 python 从 hdfs 访问文件，使用外部包，因为 open('hdfs://file') 只会引发 IOError。而且，这样一来，mesos和spark就无法正确分配job，让机器有文件处理数据，因此比使用sc.textFile()读取正常文件要慢。
公平积分。但是，在 Spark 中阅读时，我看不到另一种过滤损坏文件的方法。预处理和删除损坏的文件将是您的另一个选择。在这样做之前，可能值得做一个计时测试，看看这种方法与 sc.textFile() 在一批已知的、未损坏的 gzip 上相比要慢多少。这可以帮助确定预处理是否值得。