pyarrow write_dataset每个分区文件限制

Question

假设我有一些数据位于现有数据集中，该数据集中没有列分区，但仍然有 200 多个文件。我想将该数据重写到配置单元分区中。数据集太大，无法将其全部打开到内存中。在本例中，数据集来自 Azure Synapse 中的 CETA，但我认为这并不重要。

我这样做：

dataset=ds.dataset(datadir, format="parquet", filesystem=abfs)
new_part = ds.partitioning(pa.schema([("nodename", pa.string())]), flavor="hive")
scanner=dataset.scanner()
ds.write_dataset(scanner, newdatarootdir, 
                    format="parquet", partitioning=new_part,
                    existing_data_behavior="overwrite_or_ignore",
                    max_partitions=2467)

在本例中，有 2467 个唯一的节点名，因此我希望有 2467 个目录，每个目录有 1 个文件。然而我得到的是 2467 个目录，每个目录有 100 多个文件，每个文件大约 10KB。如何为每个分区获取 1 个文件？

我可以做第二步

for node in nodes:
    fullpath=f"{datadir}\\{node}"
    curnodeds=ds.dataset(fullpath, format="parquet")
    curtable=curnodeds.to_table()
    os.makedirs(f"{newbase}\\{node}")
    pq.write_table(curtable, f"{newbase}\\{node}\\part-0.parquet",
                version="2.6", flavor="spark", data_page_version="2.0")

有没有办法将第二步合并到第一步中？

编辑：

这是使用 pyarrow 7.0.0

编辑（2）：

使用 max_open_files=3000 确实导致每个分区一个文件。两者之间的元数据比较是，两步方法（对于一个分区）...

<pyarrow._parquet.FileMetaData object at 0x000001C7FA414A40>
created_by: parquet-cpp-arrow version 7.0.0
num_columns: 8
num_rows: 24840
num_row_groups: 1
format_version: 2.6
serialized_size: 1673

size on disk: 278kb

和一步...

<pyarrow._parquet.FileMetaData object at 0x000001C7FA414A90>
created_by: parquet-cpp-arrow version 7.0.0
num_columns: 8
num_rows: 24840
num_row_groups: 188
format_version: 1.0
serialized_size: 148313

size on disk: 1.04MB

显然，在两步版本中，我明确将版本设置为 2.6，以便解释这一差异。第 2 步后的数据总大小为 655MB，第 1 步后数据总大小为 2.6GB。时间差异也相当大。两步大约是第一步20分钟，第二步40分钟。整个过程只需要 5 分钟。

剩下的问题是，如何在write_dataset中设置version="2.6"和data_page_version="2.0"？我仍然想知道为什么 row_groups 如此不同，如果它们在设置这些参数时如此不同，但我会推迟这个问题。

Answer 1

数据集作者在7.0.0中显着更改。以前，它始终每个分区创建1个文件。现在有几个设置可能导致其编写多个文件。此外，看来您最终要获得许多小排组，这是不理想的，这可能是一步过程既较慢又较大的原因。

第一个重要的设置是max_open_files。一些系统限制了一次可以打开多少个文件描述符。 Linux默认为1024，因此Pyarrow尝试默认为〜900（假设某些文件描述符将打开以进行扫描等）。对于某些数据集，这效果很好。但是，如果每个批次都有每个文件的数据，则根本无法正常工作。在这种情况下，您可能想增加max_open_files比您的分区数大（带有一些摆动室，因为您也将有一些可以打开的文件以读取）。您可能需要调整特定于OS的设置以允许此设置（通常，这些OS限制非常保守，并且提高此限制是无害的）。

我仍然想知道为什么row_groups在设置这些参数时如此不同，但我会推迟这个问题。

此外，7.0.0版本添加min_rows_per_group，max_rows_per_group和max_rows_per_per_file参数参数到write> write_dataset呼叫。设置min_rows_per_group到100万之类的东西将导致作者在内存中缓冲行，直到足够写入为止。这将使您可以使用1行组而不是188行组创建文件。这应该降低您的文件大小并解决您的性能问题。

但是，与此相关的内存成本将为min_rows_per_group * num_files * size_of_row_in_bytes。

剩下的问题是，如何设置版本=“ 2.6”和data_page_version =“ 2.0” write_dataset？

write_dataset呼叫以几种不同格式（例如CSV，IPC，ORC）起作用，因此调用仅具有适用于格式的通用选项。

可以使用file_options参数设置格式特定的设置：

# It does not appear to be documented but make_write_options
# should accept most of the kwargs that write_table does
file_options = ds.ParquetFileFormat().make_write_options(version='2.6', data_page_version='2.0')
ds.write_dataset(..., file_options=file_options)