Apache Arrow Python使用 PyArrow 读写 Parquet 分区数据集多文件的完整实践【免费下载链接】arrowApache Arrow is the universal columnar format and multi-language toolbox for fast data interchange and in-memory analytics项目地址: https://gitcode.com/GitHub_Trending/arrow3/arrow本文基于 PyArrow 官方文档 Partitioned Datasets (Multiple Files)系统讲解如何把多个 Parquet 文件组织为分区数据集partitioned dataset包括用pq.write_to_dataset按分区列写入本地或远程文件系统、手工生成_metadata/_common_metadata元数据文件以提升读取效率、用ParquetDataset/pq.read_table自动发现并读取 Hive 风格目录分区。读完本文你可以独立完成一个可被 Spark、Dask 等框架消费的 Parquet 分区数据集的写入、元数据管理与读取全流程并能理解每个参数在源码中的实际行为。什么是 Parquet 数据集Partitioned Dataset多个 Parquet 文件共同构成一个 Parquet数据集dataset。它可以有两种形态一组 Parquet 绝对文件路径的列表一个包含嵌套目录的目录名其中目录结构本身定义了数据集的分区方式。例如一个按year、month两级分区的数据集在磁盘上的典型布局如下这也是 Spark、Hive 生态中广泛采用的 Hive 风格分区目录约定dataset_name/ year2007/ month01/ 0.parq 1.parq ... month02/ 0.parq 1.parq ... month03/ ... year2008/ month01/ ... ...关键点在于分区信息不保存在 Parquet 文件内部而是编码在目录名列名值中。读取时 PyArrow 会扫描目录、把分区列还原为表中的普通列。写入分区数据集pq.write_to_datasetpyarrow.parquet模块对任何 file-store 类型的文件系统本地、HDFS、S3 等都支持写入分区数据集不指定文件系统时默认使用本地文件系统。本地数据集写入 import pyarrow as pa import pyarrow.parquet as pq table pa.table({one: [-1, None, 2.5], ... two: [foo, bar, baz], ... three: [True, False, True]}) ... # Local dataset write pq.write_to_dataset(table, root_pathdataset_name, ... partition_cols[one, two])参数语义root_path指定数据集保存的父目录partition_cols按哪些列分区。列会按给定顺序依次分区分区切分点由分区列中的唯一值决定上例会生成one-1/twofoo、one-1/twobar、one2.5/twobaz、twobazone 为 null等组合目录。远程文件系统写入要使用其他文件系统只需传入filesystem参数。由于底层对每个分区的表写入都是用with语句包装的即用后即关闭pq.write_to_dataset调用本身无需再包在with块中 # Remote file-system example from pyarrow.fs import HadoopFileSystem # doctest: SKIP fs HadoopFileSystem(host, port, useruser, kerb_ticketticket_cache_path) # doctest: SKIP pq.write_to_dataset(table, root_pathdataset_name, # doctest: SKIP ... partition_cols[one, two], filesystemfs)兼容性提醒如果写入的数据集后续要交给 HIVE 使用分区列的取值必须与你所运行 HIVE 版本允许的目录名字符集兼容例如不能包含某些特殊字符。源码视角write_to_dataset 的完整参数在 core.py 中可以看到write_to_dataset的完整签名它本质上是dataset.write_dataset的包装器def write_to_dataset(table, root_path, partition_colsNone, filesystemNone, schemaNone, partitioningNone, basename_templateNone, use_threadsNone, file_visitorNone, existing_data_behaviorNone, **kwargs):文档示例未展示但源码明确支持的几个实用参数参数说明依据 源码 docstringschema显式指定数据集 Schema不从数据推断partitioning分区方案可传pyarrow.dataset.partitioning()对象或字段名列表用于切换 Hive 风格与目录风格partitioning_flavorbasename_template数据文件名模板{i}会被递增整数替换默认guid-{i}.parquet因此每次重写会生成新文件名use_threads默认True并行写各分区文件并行度由 CPU 核数决定file_visitor每写一个文件回调一次回调对象含path与metadataParquet 文件元数据且已设置相对文件路径可用于自行构建_metadata文件existing_data_behavioroverwrite_or_ignore默认/error/delete_matching控制对目标中已有数据的处理策略其中existing_data_behavioroverwrite_or_ignore配合每次写入唯一化的basename_template可以实现追加append工作流同名文件被覆盖、其他已有文件被忽略新分区的文件则以新名字写入。写入_metadata与_common_metadata元数据文件一些处理框架如 Spark、Dask在可选地使用分区数据集时会读取_metadata和_common_metadata这两个文件_common_metadata包含整个数据集的 Schema 信息_metadata除 Schema 外还包含所有文件中全部 row group 的元数据。这两个文件本身都是只含元数据的 Parquet 文件。需要强调这不是 Parquet 标准的一部分而是这些框架在实践中形成的约定。使用这两个文件可以显著提升 Parquet Dataset 的构建效率读取方可以直接复用已存储的 Schema 与所有 row group 的文件路径而不必推断 Schema、也不必遍历目录发现所有 Parquet 文件——在文件访问代价高昂的文件系统上如对象存储收益尤其明显。pq.write_to_dataset本身不会自动写入这类元数据文件但提供了收集元数据的机制由你手动合并写出。方式一借助 write_to_dataset 收集元数据 # Write a dataset and collect metadata information of all written files metadata_collector [] root_path dataset_name_1 pq.write_to_dataset(table, root_path, metadata_collectormetadata_collector) # Write the _common_metadata parquet file without row groups statistics pq.write_metadata(table.schema, root_path /_common_metadata) # Write the _metadata parquet file with row groups statistics of all files pq.write_metadata( ... table.schema, root_path /_metadata, ... metadata_collectormetadata_collector ... )pq.write_metadata的实现见 core.py它先用ParquetWriter把 Schema以及收集到的 FileMetaData写成只含元数据的 Parquet 文件当传入metadata_collector时源码会先落盘再回读元数据把收集器中各文件的 row group 元数据合并进该文件——这就是写入后回读的实现细节。方式二用 write_table / ParquetWriter 手工写分区文件时收集当你不用write_to_dataset而是用pq.write_table或ParquetWriter逐个写分区文件时metadata_collector关键字同样可以收集每个文件的 FileMetaData。此时有两条约束来自官方文档的明确要求你需要自己为 row group 元数据中设置的文件路径相对于分区数据集根目录赋值所有文件的 Schema 以及收集到的各 FileMetaData 对象的 Schema 必须一致。 import os os.mkdir(year2017) metadata_collector [] pq.write_table( ... table, year2017/data1.parquet, ... metadata_collectormetadata_collector ... ) # set the file path relative to the root of the partitioned dataset metadata_collector[-1].set_file_path(year2017/data1.parquet) # combine and write the metadata metadata metadata_collector[0] for _meta in metadata_collector[1:]: ... metadata.append_row_groups(_meta) metadata.write_metadata_file(_metadata) # or use pq.write_metadata to combine and write in a single step pq.write_metadata( ... table.schema, _metadata, ... metadata_collectormetadata_collector ... ) pq.read_metadata(_metadata) pyarrow._parquet.FileMetaData object at ... created_by: parquet-cpp-arrow version ... num_columns: 3 num_rows: 3 num_row_groups: 1 format_version: 2.6 serialized_size: ...FileMetaData.write_metadata_file的 Python 绑定位于 core.py。对应的行为测试分别位于 test_dataset.py 和 test_metadata.py可作为元数据收集、合并与文件路径设置的验证参考。读取分区数据集ParquetDataset 与 read_tableParquetDataset类接受一个目录名或文件路径列表能够自动发现并推断若干常见的分区结构例如 Hive 生成的目录风格分区。其类定义与参数文档见 core.py。 dataset pq.ParquetDataset(dataset_name/) table dataset.read() table pyarrow.Table three: bool one: dictionaryvaluesstring, indicesint32, ordered0 two: dictionaryvaluesstring, indicesint32, ordered0 ---- three: [[true],[true],[false]] one: [ -- dictionary: [-1,2.5] -- indices: [0], -- dictionary: [-1,2.5] -- indices: [1], -- dictionary: [null] -- indices: [0]] two: [ -- dictionary: [foo,baz,bar] -- indices: [0], -- dictionary: [foo,baz,bar] -- indices: [1], -- dictionary: [foo,baz,bar] -- indices: [2]]也可以直接用pyarrow.parquet暴露的便捷函数read_table省去显式创建 Dataset 对象的一步 table pq.read_table(dataset_name)从源码看core.pyread_table内部就是构造ParquetDataset再读取当pyarrow.dataset模块不可用时会退回到单文件的ParquetFile路径此时filters关键字将不可用并抛出ValueError。读取时的三个重要行为细节1. 分区列的类型与顺序。读取时原表中的分区列的类型会被转换为 Arrowdictionary 类型对应 pandas 的 categorical。此外分区列的顺序不会在保存/读取过程中保持。如果从远程文件系统读入 pandas DataFrame 且写入时启用了preserve_index选项可能需要执行sort_index来恢复行序。2. 比分区键更强大的过滤能力。ParquetDataset支持在所有列上过滤利用 row group 统计信息而不仅仅局限于分区键同时支持比 Hive 风格更细粒化的目录分区方案例如/2019/11/15/而非/year2019/month11/day15/并且可以为分区键指定 Schema。过滤语义在 源码参数文档 中说明嵌入在嵌套目录中的分区键会被利用若某文件不可能包含匹配行则根本不会加载该文件。3. 选取列子集时必须显式带上分区键。当你只想读取列子集时如果希望结果中包含分区键列必须把它们显式写进columns关键字# 想要拿到分区列 year 时columns 中必须显式包含它 pq.read_table(dataset_name/, columns[year, three])4. 单文件路径不会推断分区列。把单个文件路径传给read_table或ParquetDataset时即使路径中包含 Hive 风格的段如year2017也不会从路径推断出分区列。要获得分区列请传入其父目录 # Doesnt include year as a column pq.read_table(dataset_name/year2017/data1.parquet) # doctest: SKIP # Includes year as a partition column pq.read_table(dataset_name/) # doctest: SKIP这个行为与_metadata/_common_metadata文件的发现机制也有关联从 源码 可以看到ParquetDataset有ignore_prefixes参数默认值为[., _]——即以.或_开头的文件在目录发现阶段会被忽略。这正是为什么你写出的_metadata、_common_metadata侧车文件不会与数据文件混淆目录发现只关注真正的 Parquet 数据文件。总结数据集读写的关键决策点场景推荐做法依据按分区列组织写入pq.write_to_dataset(table, root_path, partition_cols[...])core.py#L2138写入 HDFS/S3 等远程 file-store传filesystemfs无需额外with包装官方文档 core.py#L2169追加写入工作流existing_data_behavioroverwrite_or_ignore 唯一basename_templatecore.py#L2208供 Spark/Dask 高效读取metadata_collector收集元数据 pq.write_metadata写出_metadata/_common_metadatacore.py#L2319整目录读取pq.read_table(dataset_name/)或ParquetDatasetcore.py#L1899读取时取分区列子集columns中必须显式包含分区键名官方文档 note掌握以上内容后你即可在本地与远程文件系统之间以统一 API 完成 Parquet 分区数据集的写入、元数据管理与读取若需了解单个 Parquet 文件的读写细节、类型映射或加密decryption_properties等参数在 read_table 与 ParquetDataset 中同样透传可继续参阅同目录下的 Parquet 读写文档 与 Parquet 类型处理文档。【免费下载链接】arrowApache Arrow is the universal columnar format and multi-language toolbox for fast data interchange and in-memory analytics项目地址: https://gitcode.com/GitHub_Trending/arrow3/arrow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考