从 Python 增量编写 Parquet 数据集答案

【问题标题】：Incrementally writing Parquet dataset from Python从 Python 增量编写 Parquet 数据集
【发布时间】：2021-07-14 09:14:07
【问题描述】：

我正在从我的 Python 应用程序中写入大于 RAM 的数据 - 基本上是将数据从 SQLAlchemy 转储到 Parque。我的解决方案受到this question 的启发。即使增加the batch size as hinted here 我也面临这些问题：

RAM 使用量大幅增长
一段时间后写入器开始变慢（写入吞吐速度下降超过 5 倍）

我的假设是，这是因为当行数增加时，ParquetWriter 元数据管理变得昂贵。我在想我应该切换到datasets，这将允许作者在处理过程中关闭文件，清除元数据。

我的问题是

有没有使用 Python 和 Parquet 编写增量数据集的示例
我的假设是正确的还是不正确的？使用数据集有助于维持写入器的吞吐量？

我的提炼代码：


writer = pq.ParquetWriter(
                    fname,
                    Candle.to_pyarrow_schema(small_candles),
                    compression='snappy',
                    allow_truncated_timestamps=True,
                    version='2.0',  # Highest available schema
                    data_page_version='2.0',  # Highest available schema
            ) as writer:

    def writeout():
        nonlocal data
        duration = time.time() - stats["started"]
        throughout = stats["candles_processed"] / duration
        logger.info("Writing Parquet table for candle %s, throughput is %s", "{:,}".format(stats["candles_processed"]), throughout)
        writer.write_table(
            pa.Table.from_pydict(
                    data,
                    writer.schema
            )
        )
        data = dict.fromkeys(data.keys(), [])
        process = psutil.Process(os.getpid())
        logger.info("Flushed %s writer, the memory usage is %s", bucket, process.memory_info())

    # Use massive yield_per() or otherwise we are leaking memory
    for item in query.yield_per(100_000):
        frame = construct_frame(row_type, item)
        for key, value in frame.items():
            data[key].append(value)

        stats["candles_processed"] += 1

        # Do regular checkopoints to avoid out of memory
        # and to log the progress to the console
        # For fine tuning Parquet writer see
        # https://issues.apache.org/jira/browse/ARROW-10052
        if stats["candles_processed"] % 100_000 == 0:
            writeout()

【问题讨论】：

1.您的吞吐量衡量提取记录、转换它们并将它们写入镶木地板所需的时间。我认为您应该尝试单独测量每个步骤以准确指出问题所在。它可能是镶木地板，但它可能是您的代码的其余部分。 2. 为什么打电话给data=dict.fromkeys(data.keys(), []) 而不是data.clear()。如果您的密钥在所有记录中都是唯一的，那么您正在泄漏内存，因为数据会不断增长。

标签： parquet pyarrow

【解决方案1】：

在这种情况下，原因是不正确地使用 Python 列表和字典作为工作缓冲区，正如 @0x26res 所指出的那样。

确保正确清除列表字典后，内存消耗问题可以忽略不计。

【讨论】：