如何解析 Foundry 中的大型压缩 csv 文件？

Question

我有一个大的 gziped csv 文件 (.csv.gz) 上传到一个大小约为 14GB 且未压缩时为 40GB 的数据集。有没有办法使用 Python Transforms 解压、读取和写入数据集，而不会导致执行程序 OOM？

Answer 1

我将在回答这个问题时协调一些策略。

首先，我想使用here 讨论的方法使用测试驱动开发来编写这个，因为我们正在处理原始文件。使用完整检查 + 构建对原始文件的迭代速度太长了，因此我将首先创建一个示例 .csv 文件并压缩它以加快开发速度。

我的示例.csv 文件如下所示：

然后我使用命令行实用程序对其进行压缩，并将其添加到我的代码存储库中，方法是将存储库克隆到我的本地计算机，将文件添加到我的开发分支，然后将结果推送回我的 Foundry 实例。

我还在我的存储库中创建了一个test 目录，因为我想确保我的解析逻辑得到正确验证。

这导致我的存储库如下所示：

专业提示：不要忘记修改您的 setup.py 和 build.gradle 文件以启用测试并专门打包您的小测试文件。

我还需要让我的解析逻辑位于我的 my_compute_function 方法之外，以便它可用于我的测试方法，所以 parse_gzip.py 看起来如下所示：

from transforms.api import transform, Output, Input
from transforms.verbs.dataframes import union_many


def read_files(spark_session, paths):
    parsed_dfs = []
    for file_name in paths:
        parsed_df = spark_session.read.option("header", "true").csv(file_name)
        parsed_dfs += [parsed_df]
    output_df = union_many(*parsed_dfs)
    return output_df


@transform(
    the_output=Output("ri.foundry.main.dataset.my-awesome-output"),
    the_input=Input("ri.foundry.main.dataset.my-awesome-input"),
)
def my_compute_function(the_input, the_output, ctx):
    session = ctx.spark_session
    input_filesystem = the_input.filesystem()
    hadoop_path = input_filesystem.hadoop_path
    files = [hadoop_path + file_status.path for file_status in input_filesystem.ls('**/*.csv.gz')]
    output_df = read_files(session, files)
    the_output.write_dataframe(output_df)

因此，我的test_gzip_csv.py 文件如下所示：

from myproject.datasets import parse_gzip
from pkg_resources import resource_filename


def test_compressed_csv(spark_session):
    file_path = resource_filename(__name__, "test.csv.gz")
    parsed_df = parse_gzip.read_files(spark_session, [file_path])
    assert parsed_df.count() == 1
    assert set(parsed_df.columns) == {"col_1", "col_2"}

重要的是在这里看到这种方法不使用.files() 调用文件系统，它使用.ls() 方法来获取文件名的迭代器。在这种情况下这是故意这样做的，因为我们不需要在 executors 中解析文件本身；我们只需要使用 Spark 的原生方法 .csv 即可使用现有功能解析压缩文件。

GZip 文件实际上是可拆分的，Spark 自己的读取这些文件的方法将是最佳的，而不是编写自己的解压缩器/.csv 解析器。如果您要尝试解压缩并手动解析它们，您将冒着 OOMing 您的工作的风险，并且需要投入更多内存以使其成功。在您也在运行的规模上，建议不要在 Python 中处理单个文件，因为它的性能无法与 Spark 相提并论。

请注意，我在这里也使用transforms.verbs.dataframes.union_many 方法来优雅地处理具有不同架构的不同文件。您可以指定“窄”、“宽”和“严格”选项来处理不同模式的情况，请参阅最适合您需求的产品文档。