当前位置：首页 > news >正文

Pandas 2.0性能优化：Arrow后端与Lazy Evaluation的工程应用

news 2026/6/9 15:18:21

Pandas 2.0性能优化：Arrow后端与Lazy Evaluation的工程应用

一、Pandas的性能天花板：内存拷贝与即时执行的代价

Pandas 是 Python 数据分析的事实标准，但在处理百万级以上的数据集时，其性能瓶颈日益凸显。核心问题有两个：一是基于 NumPy 的列存储导致大量内存拷贝（每次操作生成新对象），二是即时执行（Eager Evaluation）模式无法进行跨操作优化。

一个典型的场景：对 1000 万行的 DataFrame 执行df.query().groupby().agg()链式操作，Pandas 会依次执行每个操作，每步都生成中间结果。而如果将三个操作合并为一个执行计划，可以避免中间结果的物化，显著减少内存占用和计算时间。

Pandas 2.0 引入了两个关键改进：基于 Apache Arrow 的ArrowDtype后端（减少类型转换开销和内存占用），以及与 PyArrow 的深度集成（支持零拷贝读取 Parquet 文件）。同时，Polars 等基于 Lazy Evaluation 的替代框架提供了另一种性能优化路径。

二、Pandas 2.0 与 Lazy Evaluation 的性能机制

flowchart TB subgraph 传统Pandas["传统 Pandas (Eager)"] direction TB P1[读取数据<br/>NumPy后端] P2[操作1: query<br/>生成中间DF1] P3[操作2: groupby<br/>生成中间DF2] P4[操作3: agg<br/>生成最终结果] P1 --> P2 --> P3 --> P4 P2 -.->|内存拷贝| M1["内存峰值: 3×原始数据"] end subgraph Pandas2Arrow["Pandas 2.0 + Arrow"] direction TB A1[读取数据<br/>Arrow后端, 零拷贝] A2[操作1: query<br/>Arrow计算] A3[操作2: groupby<br/>Arrow计算] A4[操作3: agg<br/>Arrow计算] A1 --> A2 --> A3 --> A4 A1 -.->|零拷贝Parquet| M2["内存峰值: 2×原始数据<br/>减少类型转换"] end subgraph LazyEval["Lazy Evaluation (Polars)"] direction TB L1[构建逻辑计划<br/>Query Graph] L2[查询优化器<br/>谓词下推+列裁剪] L3[生成物理计划<br/>单次扫描执行] L4[执行并返回结果] L1 --> L2 --> L3 --> L4 L3 -.->|无中间物化| M3["内存峰值: 1.2×原始数据<br/>单次扫描"] end

关键机制差异：

Arrow 后端：Apache Arrow 提供了跨语言的列式内存格式，Pandas 2.0 通过ArrowDtype直接使用 Arrow 列作为存储后端。优势包括：零拷贝读取 Parquet 文件（Arrow 和 Parquet 格式兼容）、原生支持字符串类型（无需 Python 对象开销）、更高效的缺失值处理。
Lazy Evaluation：Polars 的核心优势。操作链不会立即执行，而是构建一个逻辑计划（Query Graph）。优化器在执行前对逻辑计划进行优化：谓词下推（将过滤操作提前到扫描阶段）、列裁剪（只读取需要的列）、操作融合（合并连续的映射操作）。
内存模型：Pandas 的每个操作都生成新的 DataFrame（Copy-on-Write 2.0 缓解了这个问题）；Polars 的 Lazy 模式只在最终执行时物化结果，中间过程不产生内存拷贝。

三、性能优化实践

3.1 Pandas 2.0 Arrow 后端

import pandas as pd import pyarrow as pa import pyarrow.parquet as pq # ===== Pandas 2.0 Arrow后端 ===== # 方式一：全局启用Arrow后端 pd.options.future.infer_string = True # 字符串使用ArrowStringDtype # 方式二：读取时指定Arrow后端 df = pd.read_parquet( "large_dataset.parquet", dtype_backend="pyarrow", # 使用Arrow类型后端 use_nullable_dtypes=True, # 使用可空类型 ) # Arrow后端的优势：字符串操作性能提升 # 传统Pandas: 字符串存储为Python对象，每个对象约50字节开销 # Arrow后端: 字符串存储为Arrow字符串列，无Python对象开销 # 内存对比 print(f"传统后端内存: {df.memory_usage(deep=True).sum() / 1e9:.2f} GB") # 转换为Arrow后端 df_arrow = df.convert_dtypes(dtype_backend="pyarrow") print(f"Arrow后端内存: {df_arrow.memory_usage(deep=True).sum() / 1e9:.2f} GB") # ===== 零拷贝Parquet读取 ===== # 传统方式：Parquet → NumPy → Pandas（两次内存拷贝） # Arrow方式：Parquet → Arrow → Pandas（零拷贝，共享内存） def read_parquet_zero_copy(path: str) -> pd.DataFrame: """零拷贝读取Parquet文件""" # 直接读取为Arrow Table table = pq.read_table(path, memory_map=True) # 内存映射 # 转换为Pandas DataFrame（零拷贝） df = table.to_pandas(types_mapper=pd.ArrowDtype) return df # ===== Copy-on-Write (CoW) ===== # Pandas 2.0的CoW机制：延迟拷贝，只在修改时才真正复制 pd.options.mode.copy_on_write = True def process_with_cow(df: pd.DataFrame) -> pd.DataFrame: """CoW模式下的数据处理""" # 以下操作不会产生内存拷贝 filtered = df[df["amount"] > 100] # 视图，非拷贝 sorted_df = filtered.sort_values("date") # 视图 # 只有真正修改数据时才触发拷贝 sorted_df["new_col"] = sorted_df["amount"] * 1.1 # 触发拷贝 return sorted_df

3.2 Polars Lazy Evaluation

import polars as pl # ===== Lazy Evaluation核心用法 ===== def analyze_with_polars(parquet_path: str) -> pl.DataFrame: """ 使用Polars Lazy模式进行数据分析 所有操作构建逻辑计划，最终collect()时一次性执行 """ result = ( pl.scan_parquet(parquet_path) # 延迟扫描，不读取数据 # 谓词下推：过滤条件在扫描时就应用，减少读取量 .filter(pl.col("amount") > 100) # 列裁剪：只选择需要的列，忽略其他列 .select([ "date", "category", "amount", "region" ]) # 分组聚合 .groupby(["category", "region"]) .agg([ pl.col("amount").sum().alias("total_amount"), pl.col("amount").mean().alias("avg_amount"), pl.col("date").max().alias("last_date"), pl.count().alias("record_count"), ]) # 排序 .sort("total_amount", descending=True) # 执行：将逻辑计划转化为物理执行 .collect(streaming=True) # streaming模式处理超大数据集 ) return result # ===== 查看优化后的执行计划 ===== def show_optimized_plan(parquet_path: str) -> str: """查看Polars优化后的执行计划""" lazy_df = ( pl.scan_parquet(parquet_path) .filter(pl.col("amount") > 100) .select(["date", "category", "amount"]) .groupby("category") .agg(pl.col("amount").sum()) ) # 查看优化前的逻辑计划 print("=== 优化前 ===") print(lazy_df.describe_plan()) # 查看优化后的物理计划 print("\n=== 优化后 ===") print(lazy_df.describe_optimized_plan()) # 优化器会做： # 1. 谓词下推：filter在scan时执行 # 2. 列裁剪：只读取date, category, amount三列 # 3. 投影下推：聚合后只保留需要的列 return lazy_df.describe_optimized_plan() # ===== 性能对比 ===== def benchmark_pandas_vs_polars( parquet_path: str, iterations: int = 5, ) -> dict: """Pandas vs Polars性能对比""" import time # Pandas (Eager) pandas_times = [] for _ in range(iterations): start = time.perf_counter() df = pd.read_parquet(parquet_path, dtype_backend="pyarrow") result = ( df[df["amount"] > 100] [["date", "category", "amount"]] .groupby("category") .agg({"amount": "sum"}) ) pandas_times.append(time.perf_counter() - start) # Polars (Lazy) polars_times = [] for _ in range(iterations): start = time.perf_counter() result = ( pl.scan_parquet(parquet_path) .filter(pl.col("amount") > 100) .select(["date", "category", "amount"]) .groupby("category") .agg(pl.col("amount").sum()) .collect() ) polars_times.append(time.perf_counter() - start) return { "pandas_mean_s": sum(pandas_times) / len(pandas_times), "polars_mean_s": sum(polars_times) / len(polars_times), "speedup": sum(pandas_times) / sum(polars_times), }

3.3 大数据集的分块处理

def process_large_parquet_chunked( parquet_path: str, output_path: str, chunk_size: int = 100_000, ) -> None: """ 分块处理超大数据集 避免一次性加载到内存 """ import pyarrow.parquet as pq parquet_file = pq.ParquetFile(parquet_file=parquet_path) writer = None for batch in parquet_file.iter_batches(batch_size=chunk_size): # 转换为Polars处理 chunk = pl.from_arrow(batch) # 处理逻辑 processed = ( chunk.lazy() .filter(pl.col("amount") > 0) .with_columns([ pl.col("date").str.strptime(pl.Date, "%Y-%m-%d"), pl.col("amount").cast(pl.Float64), ]) .collect() ) # 写入输出文件 if writer is None: writer = pq.ParquetWriter(output_path, processed.to_arrow().schema) writer.write_batch(processed.to_arrow()) if writer: writer.close()