parquet存储格式

parquet是一种列存储格式, 是 Apache 开源的一种列式存储文件格式主要用在大数据和机器学习的数据处理里。Spark、Pandas、PyArrow、DuckDB、Hugging Face Datasets 都能直接读写它。1. Intro**核心特点按列存储。**CSV、JSON 这类格式是一行一行存的Parquet 则把同一列的数据放在一起存。比如一张表有 id、text、score 三列行式存储的顺序是(1, hello, 0.9), (2, world, 0.8), ...列式存储的顺序是[1, 2, ...] [hello, world, ...] [0.9, 0.8, ...]列式存储的好处只读需要的列如果只要 score就不用把整列 text 从磁盘读出来I/O 大幅减少。压缩率高同一列的类型相同、值也相似配合字典编码、游程编码RLE再加上 Snappy、Zstd 等压缩算法文件通常比 CSV 小好几倍。自带 schema文件里记录了每列的类型比如 int64、string 或嵌套结构读的时候不用猜类型。可以跳过无关数据文件分成多个行组row group每个行组都存有统计信息比如每列的最小值和最大值。按条件过滤时可以直接跳过不可能有匹配行的行组。适合分布式处理一个大数据集通常切成很多个 Parquet 分片多个 worker 可以并行读取。缺点不适合频繁地单行更新或追加写入因为文件写完基本就不再改动。二进制格式不能像 CSV 那样直接用文本编辑器查看。2. row groups, column chunk和pageParquet 文件的结构分四层从外到内是文件File→ 行组Row Group→ 列块Column Chunk→ 页PageParquet 文件├── 魔数 “PAR1”├── Row Group 0│ ├── Column Chunk: id → Page, Page, …│ ├── Column Chunk: text → Page, Page, …│ └── Column Chunk: score → Page, Page, …├── Row Group 1│ └── …├── Footer元数据schema、各行组和列块的位置与统计信息└── 魔数 “PAR1”Row Group行组先把整张表按行横向切开。比如 1000 万行的数据可以分成 10 个行组每个row group 100 万行。row group内部再按列存储每一列的数据叫一个列块Column Chunk。所以 Parquet 实际上是先row group分组group内再按column存。这样既有列式存储的好处又不需要一次把整列都读进内存。row group的作用最小的并行和读取单位Spark、PyArrow 这类引擎通常按row group分配任务一个 worker 读一个或几个row group。用统计信息跳过数据footer 里记录了每个row group中每个column chunk的最小值、最大值和 null 数量。比如查询 score 0.95 时如果某个row group里 score 的最大值只有 0.8整个row group都可以直接跳过。决定内存占用写文件时一个row group的数据要先在内存里攒满才能写出去读的时候通常也是一次处理一个行组。row group大小怎么选太大比如几个 GB会导致读写内存占用高并行度也低。太小会让元数据变多、压缩率下降读取时随机 I/O 也更多。常见做法是让每个row group在 128MB 到 1GB 之间或者在几十万到一百万行左右。PyArrow 用 row_group_size 参数按行数控制。2.1. Page页每个column chunk又被切成若干页。页是编码和压缩的最小单位默认大小约 1MB。常见的页类型Data Page数据页存实际的列值以及处理 null 和嵌套结构用的 repetition/definition levels。Dictionary Page字典页如果某列用了字典编码列块开头会有一个字典页记录编号 → 真实值的映射。后面的数据页只存编号对重复值多的列比如 language、source压缩效果特别好。page的作用编码和压缩以页为单位读取时可以逐页解压不用一次解压整个列块。更细粒度的跳读较新版本的 Parquet 支持 Page Index记录了每个页的最小值和最大值。这样过滤时不仅能跳过行组还能在行组内部跳过页。各层职责层级怎么切分主要作用Row Group按行横切并行读取、按统计信息跳过、控制内存Column Chunk行组内按列只读需要的列Page列块内再切编码和压缩、更细粒度的跳读Footer文件末尾存 schema 和所有位置、统计信息读文件时最先读它meta data放在文件末尾是因为写文件时要等所有数据写完才知道每一块的位置和统计信息。所以读 Parquet 时会先读取文件结尾的 footer再按需定位到具体的行组和列块。Write a few rows to Parquet, then read them back in several ways.importpyarrowaspaimportpyarrow.computeaspcimportpyarrow.parquetaspq PATHdemo.parquetNUM_ROWS10ROW_GROUP_SIZE4# 1. 构造数据一列整数、一列文本、一列浮点、一列重复值多的类别rows[{id:i,text:fdocument number{i}lorem ipsum *(i1),score:round(i/NUM_ROWS,2),source:[web,books,code][i%3],}foriinrange(NUM_ROWS)]tablepa.Table.from_pylist(rows)print( 原始 schema )print(table.schema,\n)# 2. 写文件每 4 行一个行组数据页上限设得很小让列块被切成多个页pq.write_table(table,PATH,row_group_sizeROW_GROUP_SIZE,data_page_size64,# 字节默认约 1MBcompressionzstd,write_statisticsTrue,write_page_indexTrue,)# 3. 读元数据先读 footerpfpq.ParquetFile(PATH)metapf.metadataprint( 文件元数据 )print(f总行数{meta.num_rows}行组数{meta.num_row_groups}列数{meta.num_columns}\n)print( 每个行组 / 列块 )forrg_idxinrange(meta.num_row_groups):rgmeta.row_group(rg_idx)print(fRow Group{rg_idx}:{rg.num_rows}行,{rg.total_byte_size}字节(未压缩))forcol_idxinrange(rg.num_columns):colrg.column(col_idx)statscol.statisticsprint(f{col.path_in_schema:7}f编码{col.encodings}压缩后{col.total_compressed_size}B fmin{stats.min!r}max{stats.max!r}f有字典页{col.has_dictionary_page})print()# 4. 全量读取print( 全量读取 )print(pq.read_table(PATH).to_pandas(),\n)# 5. 只读一个行组的部分列print( 只读 Row Group 1 的 id 和 source )print(pf.read_row_group(1,columns[id,source]).to_pandas(),\n)# 6. 流式读取每批 3 行print( iter_batches(batch_size3) )forbatchinpf.iter_batches(batch_size3,columns[id,score]):print(f batch: id{batch.column(id).to_pylist()}score{batch.column(score).to_pylist()})print()# 7. 带过滤条件读取score 0.7 时Row Group 0 (max0.3) 可以直接跳过print( 过滤读取 score 0.7 )filteredpq.read_table(PATH,columns[id,score],filters[(score,,0.7)])print(filtered.to_pandas(),\n)# 8. 校验读回的数据与原始数据一致roundtrippq.read_table(PATH)assertroundtrip.equals(table),读回的数据和原始数据不一致assertpc.sum(roundtrip[id]).as_py()sum(range(NUM_ROWS))print(往返校验通过读回的数据与写入的完全一致)3. 常见用法3.1. 创建arrow arrayimportpyarrowaspaimportpyarrow.featherasfeatherimportpyarrow.parquetaspq arrpa.array([1,2,3,4,5])print(arr)输出[ 1, 2, 3, 4, 5 ]创建不同类型的array,arr_floatpa.array([1.0,2.5,3.0])arr_strpa.array([a,b,c])print(arr_float,arr_str)输出[ 1, 2.5, 3.0 ] [ a, b, c ]创建arrow table:tablepa.table({numbers:arr_float,letters:arr_str})print(table)输出:pyarrow.Table numbers: double letters: string ---- numbers: [[1,2.5,3]] letters: [[a,b,c]]3.2. 取数据, 做slice取列print(table[numbers])[ [ 1, 2.5, 3 ] ]取行print(table.slice(0,2))pyarrow.Table numbers: double letters: string ---- numbers: [[1,2.5]] letters: [[a,b]]3.3. 转pandas dataframeimportpandasaspd dftable.to_pandas()print(df)numbers letters 0 1.0 a 1 2.5 b 2 3.0 c3.4. 读写pyarrow文件pyarrow有2种文件格式:parquet (.parquet)Arrow IPC (.feather)Arrow IPC 和 Parquet 的核心区别在于Parquet 是为“长久存储和高压缩比”而设计的磁盘/归档格式Arrow IPC 是为“内存计算和进程间高速传输零拷贝”而设计的格式。两者在现代数据分析栈中通常是配合使用的关系即用 Parquet 存盘读入内存时直接转为 Arrow 格式进行计算。核心对比维度维度Arrow IPC (Inter-Process Communication)Apache Parquet主要定位内存布局的跨进程传输与共享格式长期存储与磁盘归档格式磁盘/内存一致性磁盘表示与内存布局完全一致磁盘表示与内存布局不同需解码/反序列化反序列化开销零拷贝 (Zero-copy)无需解码支持 mmap 内存映射即用即加载有解码开销必须解压、解编码并构建内存结构后才能计算压缩率与文件体积未压缩或仅轻微压缩如字典编码体积通常较大采用先进的列式压缩与编码算法体积小得多适用阶段Data in flight传输中 / 内存计算 / 进程间通信Data at rest静止状态 / 磁盘 / 长期持久化详细对比解析Apache Parquet专为“存盘”而生• 核心思想空间效率至上。Parquet 采用了高度优化的列式压缩、字典编码、RLE行程长度编码等技术把数据压缩得很小非常适合保存在硬盘上、对象存储如 S3或长期归档。• 代价当程序要读取 Parquet 数据进行计算时不能直接拿来就用必须经历解压、反序列化、组装成内存对象的过程这会消耗 CPU 和时间。• 了解更多可以参考 Apache Arrow FAQ 或 Apache Arrow 专栏解析。Arrow IPC专为“传输与计算”而生• 核心思想速度与零拷贝。Arrow IPC 的磁盘/传输格式与内存中的列式布局完全一致。• 优势• 零拷贝 (Zero-copy)接收方拿到数据后不需要反序列化直接把字节映射为内存指针即可使用。• 内存映射 (mmap)可以通过 mmap 直接打开一个 Arrow IPC 文件操作系统按需加载处理超大文件时进程内存开销极低。• IPC 类型分为 Stream 格式适合流式管道、顺序读写和 File 格式支持带 Footer 偏移量的随机访问。• 代价没有 Parquet 那么极致的压缩率文件通常比 Parquet 大。协作关系典型工作流在现代数据工程如 DuckDB、Pandas、Spark、PyArrow 等中两者的经典配合模式是静态存储数据以 Parquet 格式保存在磁盘或云端节省存储成本和网络传输带宽。内存计算读取 Parquet 文件时底层引擎直接将其解码/转换为 Arrow 内存格式例如 Python 中使用 pyarrow.parquet.read_table。高速交互/传输在不同进程、不同语言如 Python 传给 C 或 Java之间交换数据时使用 Arrow IPC 或 Arrow Flight 实现零拷贝极速传输。# writefeather.write_feather(table,table.feather)# read, 返回dataframetablefeather.read_feather(table.feather)print(table)# print(type(df)) # class pandas.core.frame.DataFrame# read, 返回arrowtablefeather.read_table(table.feather)print(table)print(type(table))# class pyarrow.lib.Tablepq.write_table(table,table.parquet)3.5. 查看schemaprint(table.schema)输出numbers: double letters: stringarrpa.array([1,None,2])print(arr)输出[ 1, null, 2 ]3.6 创建一个多个chunk的arraychunked_arrpa.chunked_array([[1,2,3],[4,5]])print(chunked_arr)print(fchunks:{chunked_arr.num_chunks})输出[ [ 1, 2, 3 ], [ 4, 5 ] ] chunks: 2往arrow table中添加数据t1pa.table({a:[1,2]})t2pa.table({a:[3,4]})combined_tablepa.concat_tables([t1,t2])print(combined_table)# pyarrow.Table# a: int64# ----# a: [[1,2],[3,4]]3.8. arrow与其他格式的zero-copy转换importnumpyasnp arrnp.array([1,2,3])pa_arrpa.array(arr)print(pa_arr)这里面没有发生内存拷贝Zero-copy零拷贝。为什么没有发生拷贝Apache Arrow 采用了精心设计的底层架构使其在处理特定的基本数据类型如整数 int、浮点数 float 等连续的原始数值类型时内存布局与 NumPy 数组完全一致。当你把这样一个一维 NumPy 数组传给 pa.array() 时PyArrow 会采用高度优化的代码路径直接让新生成的 pa_arr 去共享并指向原生 arr 的底层内存缓冲区Memory Buffer。这使得创建速度接近瞬时且几乎没有额外的内存开销。注意有些情况下会触发内存拷贝虽然你现在的代码是零拷贝但在以下几种 NumPy 转 Arrow 的常见场景中依然会触发内存拷贝包含缺失值NaN 或 None• NumPy 中的浮点数通常用 np.nan 表示缺失但没有原生的整数缺失值通常会转为 float。• Arrow 拥有独立的、基于位图Bitmap的 Null 值标记区域。如果 NumPy 数组中存在需要处理的 Null/NaN为了构建这层位图或调整对齐通常会产生拷贝。非连续的内存块Strided / Sliced Arrays• 如果你的 NumPy 数组经过了切片操作例如 arr base_arr[::2] 隔行取值其内存不再是连续的。由于 Arrow 数组要求底层数据必须完全连续此时 PyArrow 会强制进行拷贝将其重新压平组合。特定的复杂数据类型• 如果 NumPy 数组包含了 Python 对象类型dtypeobject、字符串类型、或某些多维结构Arrow 必须重新对齐和计算偏移量此时同样会引发内存拷贝。3.9. arrow to numpyimportnumpyasnp arrnp.array([1,2,3])pa_arrpa.array(arr)print(pa_arr)pa_arr.to_numpy()pyarrow不支持从nested list创建. 需要chunked_array才可以3.9 用schema创建schpa.schema([(a,pa.int64()),(b,pa.string())])table_with_schemapa.table([pa.array([1,2]),pa.array([a,b])],schemasch)print(table_with_schema)withpa.OSFile(file.parquet,rb)ass:tablepa.feather.read_table(s)print(table.shape)4. pyarrow数据处理创建数据# 1. Create sample data as Arrow Tableidspa.array([1,2,3,4,5])valuespa.array([5.2,6.3,7.1,8.4,9.9])scorespa.array([50,60,70,80,90])mini_tablepa.table({id:ids,value:values,score:scores})print(Sample data:)print(mini_table)Sample data: pyarrow.Table id: int64 value: double score: int64 ---- id: [[1,2,3,4,5]] value: [[5.2,6.3,7.1,8.4,9.9]] score: [[50,60,70,80,90]]过滤:# 2. Filter rows: Select rows with score greater than 60maskpa.compute.greater(mini_table[score],60)filteredmini_table.filter(mask)print(Rows with score 60:)print(filtered)Rows with score 60: pyarrow.Table id: int64 value: double score: int64 ---- id: [[3,4,5]] value: [[7.1,8.4,9.9]] score: [[70,80,90]]保存为文件# 3. Save and reload the filtered Table as Parquetpq.write_table(filtered,filtered.parquet)filtered_loadedpq.read_table(filtered.parquet)print(Reloaded filtered table:)print(filtered_loaded)