使用 Spark 的 Gzip 文件答案

【问题标题】：Gzip files with Spark使用 Spark 的 Gzip 文件
【发布时间】：2015-03-02 07:21:52
【问题描述】：

我有一个 Spark 作业，它需要数千个文件作为输入并从 Amazon S3 下载它们并在映射阶段处理它们，其中每个映射步骤都返回一个字符串。我想将输出压缩到 .tar.gz 文件，然后将其上传到 S3。一种方法是

outputs = sc.map(filenames).collect()
for output in outputs:
    with tempfile.NamedTemporaryFile() as tar_temp:
        tar = tarfile.open(tar_temp.name, "w:gz")
        for output in outputs:
            with tempfile.NamedTemporaryFile() as output_temp:
                output_temp.write(output)
                tar.add(output_temp.name)
        tar.close()

问题是outputs 不适合内存（但它们适合磁盘）。有没有办法在映射阶段将输出保存到主文件系统？或者也许使用循环for output in outputs 作为生成器，这样我就不必将所有内容加载到内存中？

【问题讨论】：

标签： python gzip apache-spark pyspark

【解决方案1】：

在 Spark 1.3.0 中，您将能够在 Python 中使用相同的 Java/Scala 方法 toLocalIterator。

拉取请求已合并：https://github.com/apache/spark/pull/4237

这里是指定的文档：

    """
    Return an iterator that contains all of the elements in this RDD.
    The iterator will consume as much memory as the largest partition in this RDD.
    >>> rdd = sc.parallelize(range(10))
    >>> [x for x in rdd.toLocalIterator()]
    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
    """

总而言之，它将允许您迭代输出，而无需将所有内容收集到驱动程序。

问候，

【讨论】：