瓶颈在哪里

ETL 性能优化最忌讳的事就是一上来到处调参数。没有找到真正的瓶颈之前，所有的优化都是瞎蒙。

先搞清楚一个基本事实：ETL 管道中最慢的环节决定了整体速度。这个瓶颈可能出现在抽取阶段（网络带宽限制）、转换阶段（CPU 算力不足）、或者加载阶段（数据库写入太慢）。

定位瓶颈的基本方法是用监控工具观察每个阶段的耗时和资源消耗。一个简单的做法是在管道中埋点记录时间：

import time
from contextlib import contextmanager


@contextmanager
def measure_stage(stage_name, logger=None):
    """记录每个阶段的耗时"""
    start = time.time()
    start_mem = get_memory_usage()
    try:
        yield
    finally:
        elapsed = time.time() - start
        end_mem = get_memory_usage()
        print(f"[{stage_name}] 耗时: {elapsed:.2f}s, "
              f"内存: {start_mem:.0f}MB → {end_mem:.0f}MB "
              f"(增量: {end_mem - start_mem:.0f}MB)")

在实际案例中，我曾见过一个团队花了两周优化 SQL 查询，把转换阶段从 2 小时降到 20 分钟，但整体 ETL 仍然跑了 3 小时——因为他们没发现瓶颈其实在加载阶段，数据库写入速度跟不上。

并行处理

并行是加速 ETL 最直接的手段。可以把一个大任务拆分成多个小任务同时执行。

数据分区并行

把数据按某个维度分成多个分区，每个分区独立处理。分区方式包括：

  • 按时间分区：每天的数据一个分区，多天的数据可以并行处理
  • 按范围分区：按 ID 范围拆分，比如 1-100 万、100 万-200 万
  • 按业务维度：按地区、产品类别等业务字段拆分
import concurrent.futures
import pandas as pd
from math import ceil


def process_partition(partition_id, start_id, end_id, source_config):
    """处理单个数据分区"""
    sql = f"""
    SELECT * FROM orders
    WHERE id BETWEEN {start_id} AND {end_id}
    """
    df = read_from_source(sql, source_config)
    df = transform(df)
    write_to_target(df, partition_id)
    return len(df)


def parallel_etl(total_records, partition_size=100000, max_workers=4):
    """并行 ETL 处理"""
    num_partitions = ceil(total_records / partition_size)
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for i in range(num_partitions):
            start_id = i * partition_size + 1
            end_id = min((i + 1) * partition_size, total_records)
            
            future = executor.submit(
                process_partition, i, start_id, end_id, source_config
            )
            futures.append(future)
        
        total_processed = 0
        for future in concurrent.futures.as_completed(futures):
            count = future.result()
            total_processed += count
            print(f"分区完成，已处理 {total_processed} 条")
    
    return total_processed

进程级并行

Python 的多线程由于 GIL 的限制，在 CPU 密集型任务上效果不好。对于数据转换这类计算密集型任务，应该使用多进程。

import multiprocessing as mp


def transform_chunk(chunk_df):
    """转换数据块（在子进程中执行）"""
    # 执行复杂计算
    chunk_df["total_with_tax"] = chunk_df["amount"] * 1.13
    chunk_df["category_group"] = chunk_df["category"].apply(categorize)
    chunk_df["delivery_days"] = (
        chunk_df["delivered_at"] - chunk_df["shipped_at"]
    ).dt.days
    return chunk_df


def parallel_transform(df, num_workers=None):
    """使用进程池进行并行转换"""
    if num_workers is None:
        num_workers = mp.cpu_count()
    
    # 分割 DataFrame
    chunks = np.array_split(df, num_workers * 2)
    
    with mp.Pool(num_workers) as pool:
        results = pool.map(transform_chunk, chunks)
    
    return pd.concat(results, ignore_index=True)


# 对比：串行 vs 并行
if __name__ == "__main__":
    import time
    
    # 生成模拟数据
    df = pd.DataFrame({"amount": range(1000000)})
    
    # 串行处理
    start = time.time()
    serial_result = transform_chunk(df)
    print(f"串行耗时: {time.time() - start:.2f}s")
    
    # 并行处理
    start = time.time()
    parallel_result = parallel_transform(df)
    print(f"并行耗时: {time.time() - start:.2f}s")

批量大小调优

批量大小（Batch Size）是 ETL 优化中最容易被忽视的参数。批量大小太小，网络往返次数太多，吞吐量上不去。批量大小太大，单批处理时间过长，内存压力大，失败后重试成本高。

批量大小不存在一个放之四海皆准的数值。需要通过实验找到最佳值。

import time
import matplotlib.pyplot as plt


def benchmark_batch_sizes(etl_func, batch_sizes, test_data):
    """测试不同批量大小的性能"""
    results = []
    
    for batch_size in batch_sizes:
        start = time.time()
        etl_func(test_data, batch_size)
        elapsed = time.time() - start
        
        throughput = len(test_data) / elapsed
        results.append({
            "batch_size": batch_size,
            "elapsed": elapsed,
            "throughput": throughput
        })
        print(f"批量大小 {batch_size}: {elapsed:.2f}s, 吞吐量 {throughput:.0f} 条/秒")
    
    return results


# 示例：数据库批量写入性能测试
def batch_insert(records, batch_size, db_config):
    """批量写入数据库"""
    conn = create_connection(db_config)
    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        sql = build_batch_insert_sql(batch)
        conn.execute(sql)
    conn.commit()


# 批量大小调优结果通常呈现倒 U 形曲线：
# 太小 → 网络开销大 → 慢
# 适中 → 吞吐量最高
# 太大 → 内存压力大 → 单笔事务时间长 → 慢

经验参考值（在大多数场景下）：

  • 数据库写入：每批 500-5000 行
  • API 调用：每批 100-1000 条
  • 文件读写：每批 1 万-10 万行
  • 消息队列：取决于消息大小，从几百到几千条

数据库读写优化

批量化写入

不要逐行 INSERT，使用批量 INSERT 或多行 VALUES：

-- ❌ 不推荐：逐行插入
INSERT INTO orders VALUES (1, 'A', 100);
INSERT INTO orders VALUES (2, 'B', 200);
INSERT INTO orders VALUES (3, 'C', 300);

-- ✅ 推荐：批量插入
INSERT INTO orders (id, name, amount) VALUES
(1, 'A', 100),
(2, 'B', 200),
(3, 'C', 300);

合理使用索引

对于 ETL 中的临时表或中间表，不建议建太多索引——写入数据时维护索引本身也是一种开销。

策略是这样的：

  • 抽取阶段：源表的索引应该是已经存在的，不要为 ETL 额外加太多索引
  • 转换阶段：中间表尽量少用索引，尤其是频繁写入的表
  • 加载阶段：目标表在大批量加载前，可以先禁用索引，加载完再重建
-- 大批量加载前禁用索引
ALTER INDEX idx_orders_date DISABLE;

-- 执行批量插入
INSERT INTO dw.orders SELECT * FROM staging.orders;

-- 加载完成后重建索引
ALTER INDEX idx_orders_date REBUILD;

避免 N+1 查询

在转换阶段访问参考数据时，很容易写出 N+1 查询：

# ❌ N+1 查询：每次循环都查询一次数据库
for order in orders:
    user = db.query(f"SELECT * FROM users WHERE id = {order['user_id']}")
    order["user_name"] = user["name"]

# ✅ 批量查询：一次查出所有用户
user_ids = list(set(o["user_id"] for o in orders))
users = db.query(f"SELECT * FROM users WHERE id IN ({','.join(user_ids)})")
user_map = {u["id"]: u["name"] for u in users}

for order in orders:
    order["user_name"] = user_map.get(order["user_id"])

内存管理

Python 的 Pandas 在处理大数据时，内存不足是最常见的问题之一。

分块读取

不要一次性把全部数据加载到内存。

import pandas as pd


def process_in_chunks(file_path, chunk_size=50000):
    """分块处理大文件"""
    results = []
    
    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        # 处理每个数据块
        processed = transform_chunk(chunk)
        results.append(processed)
        # 及时释放内存
        del chunk, processed
    
    return pd.concat(results, ignore_index=True)

选择合适的数据类型

Pandas 默认的数据类型往往比实际需要的类型占用更多内存。手动优化类型可以大幅降低内存使用。

def optimize_dtypes(df):
    """优化 DataFrame 数据类型以减少内存"""
    start_mem = df.memory_usage(deep=True).sum() / 1024 ** 2
    
    for col in df.columns:
        col_type = df[col].dtype
        
        if col_type != "object":
            c_min = df[col].min()
            c_max = df[col].max()
            
            if str(col_type)[:3] == "int":
                # 整数类型优化
                if c_min >= 0:
                    if c_max <= 255:
                        df[col] = df[col].astype("uint8")
                    elif c_max <= 65535:
                        df[col] = df[col].astype("uint16")
                    elif c_max <= 4294967295:
                        df[col] = df[col].astype("uint32")
                else:
                    if c_min >= -128 and c_max <= 127:
                        df[col] = df[col].astype("int8")
                    elif c_min >= -32768 and c_max <= 32767:
                        df[col] = df[col].astype("int16")
            else:
                # 浮点类型优化
                df[col] = pd.to_numeric(df[col], downcast="float")
        else:
            # 分类数据优化
            if df[col].nunique() / len(df) < 0.5:
                df[col] = df[col].astype("category")
    
    end_mem = df.memory_usage(deep=True).sum() / 1024 ** 2
    print(f"内存占用: {start_mem:.2f} MB → {end_mem:.2f} MB (减少 {100 * (start_mem - end_mem) / start_mem:.1f}%)")
    
    return df

及时清理中间结果

ETL 管道中会产生大量中间数据，如果不及时清理，内存会被慢慢吃光。

def etl_with_memory_discipline():
    """有内存纪律的 ETL"""
    # 1. 抽取 - 只保留需要的列
    raw = read_source(["order_id", "amount", "status"])
    
    # 2. 完成一个阶段后，立即释放不再需要的变量
    extracted = raw[raw["status"].notna()].copy()
    del raw  # 原始数据不再需要
    
    # 3. 转换后释放中间结果
    transformed = expensive_transform(extracted)
    del extracted
    
    # 4. 最终写入后释放全部
    write_to_target(transformed)
    del transformed
    
    # 5. 强制垃圾回收
    import gc
    gc.collect()

列式文件格式

在数据读取和传输阶段，文件格式的选择对性能影响巨大。列式存储格式（Parquet、ORC）比行式格式（CSV、JSON）在分析场景下有明显的性能优势。

特性 CSV JSON Parquet
压缩比 低 低 高（通常压缩 3-5 倍）
读取速度 慢（文本解析） 慢（JSON 解析） 快（列式裁剪）
Schema 无 隐式 显式
支持复杂类型 否 是 是
适用场景 小数据、人类可读 API 数据交换 分析型工作负载 
# 读取 Parquet 比读取 CSV 快很多
import pandas as pd

# CSV：需要解析每一行，确定列类型，占用大量 IO
df_csv = pd.read_csv("orders.csv")  # 100M 文件 → 15 秒

# Parquet：列式存储+压缩，只读取需要的列
df_parquet = pd.read_parquet("orders.parquet")  # 20M 文件 → 2 秒

# 只读取需要的列进一步加速
df_filtered = pd.read_parquet(
    "orders.parquet",
    columns=["order_id", "amount", "status"]
)

Pandas 操作优化

Pandas 是 Python ETL 中最常用的库，但默认写法不一定是最快的。

避免在循环中使用 apply

# ❌ 慢：使用 apply 逐行计算
df["total"] = df.apply(lambda row: row["price"] * row["quantity"], axis=1)

# ✅ 快：使用向量化操作
df["total"] = df["price"] * df["quantity"]

# ❌ 慢：字符串拼接用 apply
df["full_name"] = df.apply(lambda r: f"{r['first_name']} {r['last_name']}", axis=1)

# ✅ 快：向量化字符串操作
df["full_name"] = df["first_name"] + " " + df["last_name"]

使用 query 和 eval

对于过滤和运算操作，Pandas 的 query 和 eval 方法会利用底层 NumExpr 加速。

# 过滤
# ❌
df_filtered = df[(df["amount"] > 100) & (df["status"] == "PAID")]

# ✅ 使用 query（在大数据集上快 20-30%）
df_filtered = df.query("amount > 100 and status == 'PAID'")

# 计算
# ❌
df["ratio"] = (df["amount"] - df["tax"]) / df["amount"]

# ✅ 使用 eval
df["ratio"] = df.eval("(amount - tax) / amount")

避免链式赋值

# ❌ 链式赋值（可能产生 SettingWithCopyWarning）
df[df["status"] == "PENDING"]["priority"] = "high"

# ✅ 使用 loc
df.loc[df["status"] == "PENDING", "priority"] = "high"

缓存策略

在 ETL 中，有一些数据是不需要每次重新计算的。缓存这些数据可以节省大量时间。

维度表的缓存

业务维度表（如产品分类、区域信息）变更频率很低，可以缓存起来重复使用。

class DimensionCache:
    """维度表缓存（每天刷新一次）"""
    
    def __init__(self, db_config, cache_ttl_hours=24):
        self.db = create_connection(db_config)
        self.cache = {}
        self.cache_ttl = timedelta(hours=cache_ttl_hours)
        self.last_refresh = {}
    
    def get_dimension(self, table_name):
        """获取维度表数据（带缓存）"""
        now = datetime.now()
        last_refresh = self.last_refresh.get(table_name)
        
        if last_refresh and (now - last_refresh) < self.cache_ttl:
            return self.cache[table_name]
        
        print(f"刷新维度表缓存: {table_name}")
        df = pd.read_sql(f"SELECT * FROM {table_name}", self.db)
        self.cache[table_name] = df
        self.last_refresh[table_name] = now
        return df

中间结果缓存

对于多个下游需要使用的中间数据，可以把结果写入临时表，需要时直接读取。

def checkpoint_to_temp_table(df, checkpoint_name, conn):
    """将中间结果写入临时表作为检查点"""
    temp_table = f"tmp_etl_{checkpoint_name}"
    df.to_sql(temp_table, conn, if_exists="replace", index=False)
    print(f"检查点已保存: {temp_table} ({len(df)} 行)")
    return temp_table


def resume_from_checkpoint(checkpoint_name, conn):
    """从检查点恢复"""
    temp_table = f"tmp_etl_{checkpoint_name}"
    if table_exists(temp_table, conn):
        return pd.read_sql(f"SELECT * FROM {temp_table}", conn)
    return None

小结

ETL 性能优化是一个系统性的工程。首先要通过埋点和监控找到真正的瓶颈，然后对症下药。并行处理是加速最有效的手段之一，但要注意分区策略和进程/线程的选择。批量大小的调优往往能带来意想不到的收益。在内存管理上，分块读取和数据类型优化可以减少不必要的资源消耗。文件格式方面，列式存储（Parquet）在分析场景下远优于 CSV 和 JSON。最后，不要轻视缓存的作用——合理缓存维度表和中间结果可以避免大量重复计算。

Summary: ETL 性能优化的系统方法论，从瓶颈定位到并行、批量、内存和文件格式全方位优化。