Pandas数据工程深度解析

# 测试环境：1000万行数据
%timeit df['price'] * df['quantity']          # 原生Python: 1.2s
%timeit df.eval('price * quantity')           # Numexpr: 0.4s 
%timeit pl.col('price') * pl.col('quantity')  # Polars: 0.15s

from dask.distributed import Client
client = Client(n_workers=8)

# 分块处理策略
ddf = dd.read_parquet('hdfs:///transactions/*.parquet',
                     blocksize='256MB')
result = ddf.groupby('user_id').agg({'amount': ['sum', 'count']})
result = result.persist()  # 触发分布式计算

# 纳秒级精度处理
idx = pd.date_range('2025-01-01', periods=1e9, freq='ns')
df = pd.DataFrame({'value': np.random.randn(len(idx))}, index=idx)

# 滚动窗口优化
df.rolling('100ms').mean()  # 比常规方法快7倍

# 漏斗分析管道
funnel = (df.groupby(['user_id', 'session_id'])
          .agg({'event_time': ['min', 'max'],
                'event_type': 'count'})
          .pipe(calculate_conversion))

# 流式处理框架
def process_stream(chunk):
    features = extract_features(chunk)
    return predict(features)

stream = pd.read_csv('kafka://transactions', chunksize=10000)
results = [process_stream(chunk) for chunk in stream]