数据管道与ETL工作流

Claude Code 工作流专题 · 自动化数据处理流程

专题：Claude Code 工作流系统学习

关键词：Claude Code, ETL, 数据管道, Airflow, Prefect, 数据清洗, 数据质量, 数据血缘, 分区

一、引言

在现代数据驱动的业务环境中，数据管道（Data Pipeline）与ETL（Extract, Transform, Load）工作流构成了数据工程的核心支柱。ETL工作流负责从各种异构数据源中提取数据，经过清洗、转换、聚合后加载到目标存储系统中，为数据分析、机器学习和业务决策提供高质量的数据基础。使用Claude Code可以显著加速ETL工作流的开发、调试和维护过程，帮助数据工程师快速构建健壮的数据管道。

数据提取 (Extract) → 数据转换 (Transform) → 数据加载 (Load)
↑ ↓
数据质量 & 数据血缘 (贯穿全流程)

本文从六个核心维度系统阐述ETL数据管道的构建方法，涵盖数据提取、数据转换、数据加载、ETL编排、数据质量检测和数据血缘追踪。每个维度均配有完整的Python代码示例，可直接用于生产实践。无论你是刚入行的数据工程师还是寻求架构优化的资深开发者，本文提供的模式和最佳实践都能帮你构建更可靠、更可维护的数据管道。

前置知识：读者应具备Python编程基础，了解关系型数据库和SQL的基本概念。对Apache Airflow或Prefect等编排工具的了解有助于理解ETL编排章节，但非必需。

二、数据提取 (Extract)

数据提取是ETL工作流的首要环节，目标是从源系统中获取原始数据。源系统可以是关系型数据库、NoSQL数据库、API接口、Web页面、本地文件或数据流。提取策略的选择直接影响后续转换和加载的效率。合理的设计应平衡提取频率、数据量和源系统负载三者的关系。

2.1 文件读取

文件读取是最基本的数据提取方式，支持CSV、JSON、Parquet、Avro等多种格式。处理大文件时应采用流式读取或分块读取策略，避免内存溢出。

import pandas as pd
import json
from pathlib import Path

# CSV文件分块读取
def read_csv_in_chunks(filepath, chunksize=10000):
    for chunk in pd.read_csv(filepath, chunksize=chunksize):
        yield chunk

# JSON文件流式读取（每行一个JSON对象）
def read_json_lines(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

# Parquet文件读取（列式存储，适合大数据量）
df = pd.read_parquet('sales_data.parquet', columns=['date', 'amount', 'region'])

2.2 数据库查询

从关系型数据库提取数据时，需要关注连接池管理、查询分页和增量抽取策略。对于生产环境，应优先使用ORM或连接池库（如SQLAlchemy、psycopg2的连接池功能）。

from sqlalchemy import create_engine, text
from sqlalchemy.pool import QueuePool

# 创建带连接池的引擎
engine = create_engine(
    'postgresql://user:pass@host:5432/db',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10
)

# 带游标的分页查询
def extract_with_cursor(engine, table, batch_size=5000):
    last_id = 0
    while True:
        query = text(f'''
            SELECT * FROM {table}
            WHERE id > :last_id
            ORDER BY id
            LIMIT :limit
        ''')
        batch = pd.read_sql(query, engine, params={
            'last_id': last_id,
            'limit': batch_size
        })
        if batch.empty:
            break
        last_id = batch['id'].iloc[-1]
        yield batch

2.3 API调用与Web爬虫

调用RESTful API时需处理认证、限流（Rate Limiting）、重试和分页。对于需要模拟浏览器行为的场景，可使用Playwright或Selenium。使用Claude Code可以快速生成适配各种API模式的客户端代码。

import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

# 异步API调用带指数退避重试
@retry(stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30))
async def fetch_api_data(session, url, params=None):
    async with session.get(url, params=params) as resp:
        resp.raise_for_status()
        return await resp.json()

async def extract_with_pagination(base_url, api_key, max_pages=100):
    headers = {'Authorization': f'Bearer {api_key}'}
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = []
        for page in range(1, max_pages + 1):
            tasks.append(fetch_api_data(
                session, base_url,
                params={'page': page, 'per_page': 100}
            ))
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [item for r in results if not isinstance(r, Exception)
                    for item in r]

2.4 流式数据与增量抽取

流式数据源（如Kafka、Kinesis）以持续不断的方式产生数据，需要消费者以至少一次或恰好一次的语义进行消费。增量抽取则通过记录时间戳或版本号，只提取自上次抽取以来发生变化的数据，显著降低传输量和源系统负载。

from kafka import KafkaConsumer
import msgpack

# Kafka流式消费
consumer = KafkaConsumer(
    'user_events',
    bootstrap_servers=['localhost:9092'],
    group_id='etl_consumer',
    value_deserializer=lambda m: msgpack.unpackb(m),
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

for msg in consumer:
    process_event(msg.value)

# 增量抽取：基于时间戳和水位线
class IncrementalExtractor:
    def __init__(self, watermark_store):
        self.watermark = watermark_store  # Redis或文件

    def extract_incremental(self, table, timestamp_col):
        last_watermark = self.watermark.get(f'watermark:{table}')
        query = f'''
            SELECT * FROM {table}
            WHERE {timestamp_col} > '{last_watermark}'
            ORDER BY {timestamp_col}
        '''
        new_data = pd.read_sql(query, engine)
        new_watermark = new_data[timestamp_col].max()
        self.watermark.set(f'watermark:{table}', new_watermark)
        return new_data

2.5 全量抽取与策略对比

抽取策略	适用场景	优点	缺点
全量抽取	数据量小、首次加载、维表	实现简单、数据完整	耗时长、资源消耗大
增量抽取	大表、频繁更新、日志表	效率高、负载低	需维护水位线、可能漏数据
CDC（变更数据捕获）	核心交易系统、实时同步	延迟低、不侵入源系统	技术复杂、成本高
流式消费	实时事件、日志、监控	准实时、扩展性好	需处理乱序和重复

三、数据转换 (Transform)

数据转换是ETL中最具业务价值也最复杂的环节。原始数据通常存在缺失值、异常值、格式不统一、重复记录等问题，需要通过一系列清洗和转换操作将其变为符合目标schema的结构化数据。Claude Code可以帮助你快速编写类型安全的转换函数和管道。

3.1 数据清洗

清洗阶段处理空值、重复值和异常值。策略包括删除、填充（均值/中位数/众数/前向填充）或标记。对于业务关键字段，建议优先采用填充而非删除。

import pandas as pd
import numpy as np

def clean_dataframe(df):
    # 删除完全重复行
    df = df.drop_duplicates()

    # 数值列用中位数填充空值
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())

    # 类别列用众数填充
    cat_cols = df.select_dtypes(include=['object']).columns
    for col in cat_cols:
        df[col] = df[col].fillna(df[col].mode().iloc[0] if not df[col].mode().empty else 'UNKNOWN')

    # 标记异常值（3 sigma规则）
    for col in numeric_cols:
        mean, std = df[col].mean(), df[col].std()
        outlier_mask = np.abs(df[col] - mean) > (3 * std)
        df.loc[outlier_mask, col] = mean

    return df

3.2 格式化与标准化

数据标准化确保不同来源的数据具有一致的表示形式，包括日期格式统一、字符串规范化、编码转换和度量单位统一。例如，中英文日期格式、电话号码格式、货币单位等都需要在转换阶段处理。

from datetime import datetime
import re

def standardize_date(date_str):
    # 支持多种输入格式
    formats = ['%Y-%m-%d', '%Y/%m/%d', '%m/%d/%Y',
               '%Y年%m月%d日', '%d-%m-%Y']
    for fmt in formats:
        try:
            return datetime.strptime(date_str.strip(), fmt).strftime('%Y-%m-%d')
        except ValueError:
            continue
    return None

def normalize_phone(phone):
    # 只保留数字
    digits = re.sub(r'\D', '', str(phone))
    if len(digits) == 11 and digits.startswith('1'):
        return digits
    return None

def standardize_column_names(df):
    # 驼峰转蛇形，去特殊字符
    rename_map = {}
    for col in df.columns:
        new_col = re.sub(r'([a-z])([A-Z])', r'\1_\2', col)
        new_col = re.sub(r'[^a-zA-Z0-9_]', '_', new_col).lower()
        rename_map[col] = new_col
    return df.rename(columns=rename_map)

3.3 聚合与连接

聚合操作将细粒度数据汇总为更高层次的分析指标，连接操作则将不同来源的数据进行关联。使用Claude Code可以快速调试复杂的多表连接逻辑，确保关联键不产生意外的笛卡尔积。

# 多维度聚合
agg_df = df.groupby(['region', 'product_category', 'date']).agg(
    total_sales=('amount', 'sum'),
    order_count=('order_id', 'nunique'),
    avg_order_value=('amount', 'mean'),
    max_single=('amount', 'max')
).reset_index()

# 多表连接（防止笛卡尔积检查）
def safe_merge(left, right, on, how='inner'):
    # 合并前检查关联键是否唯一
    left_dups = left[on].duplicated().any()
    right_dups = right[on].duplicated().any()
    if left_dups or right_dups:
        print(f'[WARN] 关联键存在重复，可能导致笛卡尔积')
    result = left.merge(right, on=on, how=how, suffixes=('_left', '_right'))
    print(f'[INFO] 合并前左表{len(left)}行，右表{len(right)}行，合并后{len(result)}行')
    return result

3.4 脱敏与类型转换

数据脱敏是安全合规的关键环节，尤其在处理PII（个人身份信息）时。常见的脱敏技术包括掩码、泛化、替换、加盐哈希和差分隐私。类型转换确保目标schema的兼容性，避免加载阶段的数据截断或类型错误。

import hashlib

# 数据脱敏函数集
class DataMasker:
    @staticmethod
    def mask_phone(phone):
        return phone[:3] + '****' + phone[7:]

    @staticmethod
    def mask_email(email):
        local, domain = email.split('@')
        return local[0] + '***@' + domain

    @staticmethod
    def hash_id(value, salt='etl_salt'):
        return hashlib.sha256(
            (str(value) + salt).encode()
        ).hexdigest()[:16]

    @staticmethod
    def generalize_age(age):
        return (age // 10) * 10

# 类型转换与字段映射
TYPE_MAPPING = {
    'source_str': str,
    'source_int': lambda x: int(x) if x else 0,
    'source_float': lambda x: float(x.replace(',', '')) if x else 0.0,
    'source_bool': lambda x: str(x).lower() in ['yes', 'true', '1', '是'],
    'source_date': standardize_date,
}

def apply_field_mapping(df, field_map):
    for target_col, (source_col, converter) in field_map.items():
        df[target_col] = df[source_col].apply(converter)
    return df

最佳实践：在转换管道的每个步骤后执行数据质量检查，尽早发现问题。使用Claude Code可以在几分钟内为转换逻辑生成对应的单元测试，极大提升代码可靠性。

四、数据加载 (Load)

数据加载将转换后的数据写入目标存储系统，目标可以是数据仓库、数据湖、搜索引擎或消息队列。加载策略的选择取决于数据量、时效性要求和目标系统的能力约束。错误的加载策略可能导致数据不一致或性能瓶颈。

4.1 批量写入与增量更新

批量写入适用于离线调度的大规模数据加载，增量更新则确保目标系统与源系统保持同步。两种模式的结合使用可以同时满足历史数据回填和实时同步的需求。

from sqlalchemy.dialects.postgresql import insert
from sqlalchemy import Table, MetaData

class DataLoader:
    def __init__(self, engine, batch_size=5000):
        self.engine = engine
        self.batch_size = batch_size

    def bulk_insert(self, table_name, df):
        # 批量写入（覆盖模式）
        df.to_sql(
            table_name,
            self.engine,
            if_exists='replace' if True else 'append',
            method='multi',
            chunksize=self.batch_size
        )

    def upsert(self, table_name, df, primary_keys):
        # PostgreSQL Upsert (INSERT ... ON CONFLICT)
        metadata = MetaData()
        table = Table(table_name, metadata, autoload_with=self.engine)
        stmt = insert(table).values(df.to_dict(orient='records'))
        stmt = stmt.on_conflict_do_update(
            index_elements=primary_keys,
            set_={c: stmt.excluded[c] for c in df.columns
                  if c not in primary_keys}
        )
        with self.engine.begin() as conn:
            conn.execute(stmt)

4.2 分区写入策略

分区是数据仓库中提升查询性能的核心技术。按时间分区尤其常见，可以快速裁剪扫描范围。合理的分区策略能将查询性能提升10倍以上。分区写入时需要确保数据路由到正确的分区目录。

from datetime import datetime

class PartitionedWriter:
    def __init__(self, base_path):
        self.base_path = base_path

    def write_partitioned(self, df, partition_col='date'):
        for (dt_val,), group in df.groupby([partition_col]):
            # 按年月日分区: /base/date=2026-05-01/
            partition_path = (
                Path(self.base_path) /
                f'{partition_col}={dt_val}'
            )
            partition_path.mkdir(parents=True, exist_ok=True)

            # 使用parquet格式写入（列式存储+压缩）
            file_path = partition_path / f'data_{datetime.now().strftime("%H%M%S")}.parquet'
            group.to_parquet(file_path, index=False, compression='snappy')
            print(f'[LOAD] 写入{len(group)}行到 {file_path}')

4.3 数据验证

加载完成后必须进行数据验证，确保数据完整性和一致性。验证应包括行数校验、汇总指标校验、空值率检查和目标端查询测试。自动化验证是CI/CD数据管道的关键环节。

import great_expectations as ge

def validate_load(df, expectations):
    # 使用Great Expectations进行数据验证
    ge_df = ge.from_pandas(df)

    results = []
    for expectation in expectations:
        result = ge_df.expect_column_values_to_not_be_null(expectation['column'])
        results.append({
            'column': expectation['column'],
            'passed': result['success'],
            'unexpected_count': result['result'].get('unexpected_count', 0)
        })

    failed = [r for r in results if not r['passed']]
    if failed:
        raise ValueError(f'数据验证失败: {failed}')

    return results

五、ETL编排 (Orchestration)

ETL编排层负责调度、监控和协调所有数据处理任务。现代编排工具（如Apache Airflow、Prefect）提供了任务依赖管理、重试机制、调度频率控制、日志跟踪和告警能力。使用Claude Code可以快速定义DAG（有向无环图）并注入自定义的业务逻辑。

5.1 Airflow DAG示例

Apache Airflow是目前最流行的开源工作流编排平台。DAG定义了任务的执行顺序和依赖关系。Claude Code可以帮助你快速编写DAG定义文件，处理复杂的依赖关系和条件分支。

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alert@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=2),
}

with DAG(
    dag_id='sales_etl_pipeline',
    default_args=default_args,
    description='销售数据ETL管道',
    schedule='0 2 * * *',  # 每天凌晨2点执行
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['etl', 'sales'],
) as dag:

    start = DummyOperator(task_id='start')

    def extract_sales(**context):
        """从API提取前一天的销售数据"""
        execution_date = context['logical_date']
        data = extract_from_api(date=execution_date)
        context['ti'].xcom_push(key='raw_data', value=data)
        return len(data)

    extract_task = PythonOperator(
        task_id='extract_sales_data',
        python_callable=extract_sales,
        provide_context=True
    )

    def choose_transform_path(**context):
        """根据数据量分支：大数据走Spark，小数据走Pandas"""
        row_count = context['ti'].xcom_pull(
            task_ids='extract_sales_data'
        )
        if row_count > 1000000:
            return 'spark_transform'
        return 'pandas_transform'

    branch = BranchPythonOperator(
        task_id='branch_on_volume',
        python_callable=choose_transform_path
    )

    def pandas_transform(**context):
        raw = context['ti'].xcom_pull(key='raw_data')
        df = pd.DataFrame(raw)
        df = clean_dataframe(df)
        context['ti'].xcom_push(key='transformed', value=df.to_json())

    pandas_task = PythonOperator(
        task_id='pandas_transform',
        python_callable=pandas_transform
    )

    # Spark转换任务（略）
    spark_task = DummyOperator(task_id='spark_transform')

    # 定义依赖关系
    start >> extract_task >> branch
    branch >> pandas_task >> load_to_warehouse
    branch >> spark_task >> load_to_warehouse

5.2 Prefect Pipeline示例

Prefect是新一代的工作流编排框架，相比Airflow提供了更优雅的Python原生API和更完善的状态管理。Claude Code可以帮助你快速将现有的Python函数装饰为Prefect任务。

from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
from prefect.task_runners import ConcurrentTaskRunner

@task(retries=2, retry_delay_seconds=30,
      cache_key_fn=task_input_hash,
      cache_expiration=timedelta(minutes=30))
def fetch_orders(date_str):
    """提取订单数据，带缓存"""
    return pd.read_sql(
        f"SELECT * FROM orders WHERE date = '{date_str}'",
        engine
    )

@task
def transform_orders(orders_df):
    """清洗和转换订单数据"""
    orders_df['total_amount'] = (
        orders_df['quantity'] * orders_df['unit_price']
    )
    orders_df = orders_df.dropna(subset=['order_id'])
    return orders_df

@task
def load_orders(transformed, table='dw.orders'):
    """使用Upsert模式加载"""
    loader = DataLoader(engine)
    loader.upsert(table, transformed, primary_keys=['order_id'])

@flow(name='order_etl', task_runner=ConcurrentTaskRunner())
def order_etl_flow(dates):
    """并行处理多天的订单数据"""
    logger = get_run_logger()
    for d in dates:
        raw = fetch_orders(d)
        logger.info(f'提取到 {len(raw)} 条订单')
        clean = transform_orders(raw)
        load_orders(clean)
        logger.info(f'完成 {d} 的数据处理')

if __name__ == '__main__':
    order_etl_flow(['2026-05-01', '2026-05-02', '2026-05-03'])

5.3 任务依赖与重试策略

合理的任务依赖设计和重试策略是生产级ETL管道的基石。任务依赖定义了管道的拓扑结构，重试策略则处理瞬态故障。Claude Code可以帮助你分析DAG的依赖关系，识别潜在的死锁和循环依赖。

重试策略设计原则：

指数退避： 首次重试等待2秒，下次4秒，再下次8秒，避免雪崩
幂等性要求： 每个任务必须可安全重试，不会产生重复数据
最大重试次数： 通常3次为宜，超过后进入人工处理流程
错误分类： 临时性错误（网络超时）= 重试，永久性错误（schema不匹配）= 立即失败
超时设置： 每个任务设置合理的execution_timeout，防止任务挂起阻塞下游

5.4 调度频率与Sensor

调度频率应根据业务数据到达的节奏设置。Airflow的Sensor（传感器）可以等待外部条件满足后再触发下游任务，非常适合处理数据依赖场景。

from airflow.sensors.external_task import ExternalTaskSensor
from airflow.sensors.filesystem import FileSensor

# 等待上游DAG完成
wait_for_datalake = ExternalTaskSensor(
    task_id='wait_for_datalake',
    external_dag_id='datalake_ingestion',
    external_task_id='complete',
    timeout=3600,  # 最大等待1小时
    poke_interval=60,  # 每60秒检查一次
    mode='reschedule'  # 等待期间释放worker slot
)

# 等待文件到达S3
wait_for_file = FileSensor(
    task_id='wait_for_data_file',
    filepath='s3://data-lake/incoming/sales_20260508.csv',
    poke_interval=30,
    timeout=1800
)

六、数据质量检测

数据质量决定了数据驱动决策的可信度。低质量数据会导致错误的业务洞察和决策风险。数据质量检测应贯穿ETL全流程，在提取、转换、加载的每个阶段都设置质量检查点。Claude Code可以协助编写和维护定制化的质量检查规则引擎。

6.1 六大质量维度

维度	定义	检查方法	阈值
完整性	数据不缺失	NULL值计数、空字符串检测	非空率 ≥ 99%
唯一性	无重复记录	主键重复检测、组合键校验	重复率 = 0%
准确性	数据反映真实值	与源系统交叉验证	准确率 ≥ 99.5%
一致性	数据格式和范围统一	格式正则校验、枚举值检查	一致率 ≥ 100%
时效性	数据及时更新	最大时间戳与系统时间比较	延迟 ≤ 1h
唯一性	无非法外键引用	引用完整性外键校验	违规行 = 0

6.2 质量检查引擎实现

from enum import Enum
from dataclasses import dataclass
from typing import Callable, Any
import pandas as pd

class Severity(Enum):
    ERROR = 'ERROR'
    WARN = 'WARN'
    INFO = 'INFO'

@dataclass
class QualityCheckResult:
    check_name: str
    passed: bool
    severity: Severity
    detail: str
    row_count: int = 0
    failed_count: int = 0

class QualityChecker:
    def __init__(self):
        self.checks = []

    def add_check(self, name, func, severity=Severity.ERROR):
        self.checks.append((name, func, severity))

    def check_not_null(self, col, threshold=0.99):
        def _check(df):
            null_count = df[col].isnull().sum()
            null_rate = null_count / len(df)
            passed = null_rate <= (1 - threshold)
            return QualityCheckResult(
                check_name=f'not_null_{col}',
                passed=passed,
                severity=Severity.ERROR,
                detail=f'{col}列空值率: {null_rate:.2%}',
                row_count=len(df),
                failed_count=null_count
            )
        return _check

    def check_unique(self, cols):
        def _check(df):
            dups = df.duplicated(subset=cols, keep=False).sum()
            passed = dups == 0
            return QualityCheckResult(
                check_name=f'unique_{"_".join(cols)}',
                passed=passed,
                severity=Severity.ERROR,
                detail=f'重复记录数: {dups}',
                row_count=len(df),
                failed_count=dups
            )
        return _check

    def check_range(self, col, min_val=None, max_val=None):
        def _check(df):
            out_of_range = pd.Series([False] * len(df))
            if min_val is not None:
                out_of_range |= df[col] < min_val
            if max_val is not None:
                out_of_range |= df[col] > max_val
            failed = out_of_range.sum()
            passed = failed == 0
            return QualityCheckResult(
                check_name=f'range_{col}',
                passed=passed,
                severity=Severity.WARN,
                detail=f'范围外: {failed}行 (期望 [{min_val}, {max_val}])',
                row_count=len(df),
                failed_count=failed
            )
        return _check

    def run_all(self, df):
        results = []
        for name, check_func, severity in self.checks:
            result = check_func(df)
            result.severity = severity
            results.append(result)
        return results

    def compute_quality_score(self, results):
        if not results:
            return 100.0
        weights = {
            Severity.ERROR: 10,
            Severity.WARN: 3,
            Severity.INFO: 1
        }
        total_weight = sum(weights[r.severity] for r in results)
        failed_weight = sum(
            weights[r.severity] for r in results if not r.passed
        )
        score = (1 - failed_weight / total_weight) * 100
        return round(score, 2)

6.3 质量评分与质量看板

数据质量评分是对整体数据健康度的量化评估，便于跟踪质量趋势。一个完善的质量看板应展示各表的质量评分趋势、各维度的通过率、Top N质量问题和高延迟的数据链路。

# 质量评分聚合
def generate_quality_report(checker, all_tables):
    report = []
    for table_name, df in all_tables:
        results = checker.run_all(df)
        score = checker.compute_quality_score(results)
        report.append({
            'table': table_name,
            'rows': len(df),
            'quality_score': score,
            'checks_total': len(results),
            'checks_failed': sum(1 for r in results if not r.passed),
        })
    return pd.DataFrame(report)

# 输出HTML质量看板片段
def render_quality_dashboard(report_df):
    html = '<table>'
    html += '<tr><th>数据表</th><th>行数</th>'
    html += '<th>质量评分</th><th>失败检查数</th></tr>'
    for _, row in report_df.iterrows():
        color = 'green' if row['quality_score'] >= 95 else (
            'orange' if row['quality_score'] >= 80 else 'red'
        )
        html += f'<tr><td>{row["table"]}</td>'
        html += f'<td>{row["rows"]}</td>'
        html += f'<td style="color:{color};font-weight:bold">{row["quality_score"]}</td>'
        html += f'<td>{row["checks_failed"]}/{row["checks_total"]}</td></tr>'
    html += '</table>'
    return html

注意：质量检查不是一次性活动。应将质量检查嵌入到CI/CD流程中，对每个ETL变更运行完整的质量回归测试。Claude Code可以辅助生成质量报告并标记回归问题。

七、数据血缘 (Data Lineage)

数据血缘（Data Lineage）记录了数据从源头到目标的完整生命周期，包括数据从哪里来、经过哪些转换、被哪些下游系统使用。血缘信息是数据治理、影响分析和故障排查的关键基础设施。Claude Code可以帮助你在项目中嵌入血缘追踪逻辑，实现自动化的元数据采集。

7.1 数据来源追踪

在数据管道的每个步骤记录来源信息，包括上游表名、提取时间戳、过滤条件和转换函数签名。这些信息为后续的审计和问题排查提供了关键线索。

from dataclasses import dataclass, field
from datetime import datetime
from uuid import uuid4

@dataclass
class LineageRecord:
    """数据血缘记录"""
    node_id: str = field(default_factory=lambda: uuid4().hex[:12])
    source_system: str = ''
    source_table: str = ''
    target_table: str = ''
    transformation: str = ''
    row_count_in: int = 0
    row_count_out: int = 0
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    execution_id: str = ''
    def to_dict(self):
        return {k: str(v) if not isinstance(v, (int, float))
                else v for k, v in self.__dict__.items()}

7.2 转换记录与依赖追踪

通过装饰器模式或上下文管理器，自动捕获每个ETL步骤的数据流信息。Claude Code可以帮助你实现无侵入的血缘采集层，与现有的ETL代码无缝集成。

from functools import wraps
import json

# 血缘采集装饰器
class LineageTracker:
    def __init__(self, output_path='lineage/'):
        self.records = []
        self.output_path = output_path

    def track(self, source, target, transform_name=''):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                result = func(*args, **kwargs)
                record = LineageRecord(
                    source_system=source,
                    source_table=source,
                    target_table=target,
                    transformation=transform_name or func.__name__,
                    row_count_in=kwargs.get('input_count', 0),
                    row_count_out=len(result) if hasattr(result, '__len__') else 0,
                    execution_id=kwargs.get('execution_id', '')
                )
                self.records.append(record)
                self._persist(record)
                return result
            return wrapper
        return decorator

    def _persist(self, record):
        path = Path(self.output_path)
        path.mkdir(exist_ok=True)
        file_path = path / f'lineage_{record.node_id}.json'
        file_path.write_text(json.dumps(record.to_dict(), ensure_ascii=False, indent=2))

    def build_lineage_graph(self):
        """构建血缘DAG"""
        graph = {}
        for record in self.records:
            source_key = f'{record.source_system}.{record.source_table}'
            target_key = f'{record.target_table}'
            if target_key not in graph:
                graph[target_key] = {'sources': set(), 'transformations': []}
            graph[target_key]['sources'].add(source_key)
            graph[target_key]['transformations'].append(record.transformation)
        return {k: {'sources': list(v['sources']),
                         'transformations': list(set(v['transformations']))}
                for k, v in graph.items()}

7.3 影响分析与审计日志

当上游数据源发生变更（如schema修改、数据回溯）时，血缘信息可以快速识别所有受影响的下游表和报表。审计日志则记录了数据管道的每一次操作，满足合规要求。这些信息通常存储在元数据仓库（如DataHub、Amundsen）中。

# 影响分析
def impact_analysis(lineage_graph, changed_table):
    """查找受影响的表和下游依赖"""
    affected = set()
    queue = [changed_table]

    while queue:
        current = queue.pop(0)
        for target, meta in lineage_graph.items():
            if current in meta['sources']:
                if target not in affected:
                    affected.add(target)
                    queue.append(target)

    return affected

# 审计日志写入
def write_audit_log(execution_id, status, details):
    log_entry = {
        'execution_id': execution_id,
        'timestamp': datetime.now().isoformat(),
        'status': status,  # SUCCESS / FAILED / RUNNING
        'details': details
    }
    with open('audit/etl_audit.log', 'a', encoding='utf-8') as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')

7.4 元数据管理

元数据管理是数据治理的核心，包括技术元数据（schema、分区信息、数据量）、业务元数据（指标定义、维度描述、数据owner）和操作元数据（运行历史、性能指标）。

# 元数据注册器
class MetadataRegistry:
    def __init__(self):
        self.catalog = {}

    def register_table(self, table_name, schema, description,
                          partition_cols=None, owner=''):
        self.catalog[table_name] = {
            'schema': schema,
            'description': description,
            'partition_cols': partition_cols or [],
            'owner': owner,
            'created_at': datetime.now().isoformat(),
            'row_count': 0,
            'last_updated': None
        }

    def update_stats(self, table_name, row_count):
        if table_name in self.catalog:
            self.catalog[table_name]['row_count'] = row_count
            self.catalog[table_name]['last_updated'] = datetime.now().isoformat()

    def search(self, keyword):
        """按关键字搜索元数据"""
        results = []
        for name, meta in self.catalog.items():
            if keyword.lower() in name.lower() or \
               keyword.lower() in meta['description'].lower():
                results.append((name, meta))
        return results

八、综合实践：端到端ETL管道

本章结合前述所有知识点，构建一个完整的ETL管道示例。该管道从多源提取数据，执行清洗转换后加载到数据仓库，同时记录数据质量和血缘信息。Claude Code在整个过程中扮演从代码生成到调试优化的助手角色。

8.1 完整管道代码

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('etl_pipeline')

class ETLPipeline:
    def __init__(self, source_engine, target_engine):
        self.source = source_engine
        self.target = target_engine
        self.quality_checker = QualityChecker()
        self.lineage_tracker = LineageTracker()
        self._setup_quality_checks()

    def _setup_quality_checks(self):
        self.quality_checker.add_check(
            'order_id_not_null',
            self.quality_checker.check_not_null('order_id')
        )
        self.quality_checker.add_check(
            'amount_in_range',
            self.quality_checker.check_range('amount', min_val=0, max_val=100000)
        )
        self.quality_checker.add_check(
            'customer_id_unique',
            self.quality_checker.check_unique(['order_id'])
        )

    def run(self, date):
        execution_id = datetime.now().strftime('%Y%m%d_%H%M%S')
        logger.info(f'[ETL] 开始执行 {execution_id}')

        # Step 1: Extract
        df = pd.read_sql(
            f"SELECT * FROM orders WHERE order_date = '{date}'",
            self.source
        )
        logger.info(f'[EXTRACT] 提取 {len(df)} 行订单数据')

        # Step 2: Quality Check (提取后)
        pre_check = self.quality_checker.run_all(df)
        self.lineage_tracker.records.append(
            LineageRecord(source_table='source.orders',
                          target_table='staging.orders_raw',
                          transformation='extract',
                          row_count_in=0, row_count_out=len(df))
        )

        # Step 3: Transform
        df = clean_dataframe(df)
        df['amount'] = df['amount'].fillna(0)
        df['etl_processed_at'] = datetime.now()

        # Step 4: Quality Check (转换后)
        post_check = self.quality_checker.run_all(df)
        quality_score = self.quality_checker.compute_quality_score(post_check)
        logger.info(f'[QUALITY] 数据质量评分: {quality_score}')

        if quality_score < 90:
            logger.error('[ETL] 质量评分过低，终止加载')
            write_audit_log(execution_id, 'FAILED',
                          {'quality_score': quality_score})
            return False

        # Step 5: Load
        loader = DataLoader(self.target)
        loader.upsert('dw.fact_orders', df, primary_keys=['order_id'])

        # Step 6: Lineage
        self.lineage_tracker.records.append(
            LineageRecord(source_table='staging.orders_raw',
                          target_table='dw.fact_orders',
                          transformation='clean_transform_load',
                          row_count_in=len(df), row_count_out=len(df))
        )

        write_audit_log(execution_id, 'SUCCESS',
                       {'rows_loaded': len(df),
                        'quality_score': quality_score})
        logger.info(f'[ETL] 执行完成，加载 {len(df)} 行')
        return True

8.2 Claude Code在ETL开发中的作用

Claude Code在整个ETL开发过程中扮演以下关键角色：

代码生成： 基于自然语言描述生成完整的ETL函数和DAG定义，大幅缩短开发周期
调试优化： 快速定位数据管道中的性能瓶颈和逻辑错误，提供优化建议
模板化： 创建可复用的ETL模块和代码模板，统一团队开发风格
测试辅助： 生成单元测试和集成测试代码，确保数据管道的可靠性
文档生成： 自动生成数据字典和管道文档，维护数据治理知识库
故障分析： 分析错误日志和数据异常，定位根因并提出修复方案

提示：在Claude Code中描述你的数据源结构和目标schema，Claude即可自动生成适配的ETL代码。对于复杂转换逻辑，可以先描述业务规则，Claude会推荐最佳实现方案。

九、核心要点总结

一、数据提取：根据数据源类型选择合适应提取策略。全量抽取适用于小表或首次加载，增量抽取适用于大表频繁更新场景，CDC提供准实时变更捕获能力。提取时注意连接池管理、分页查询和异常重试。

二、数据转换：清洗阶段处理空值、重复值和异常值；标准化阶段统一格式和编码；聚合和连接阶段需要关注数据完整性和连接键的唯一性。脱敏处理是合规刚需，应在转换阶段执行。

三、数据加载：批量写入适用于离线调度，Upsert模式保持目标数据最新。分区写入显著提升查询性能，数据验证确保加载结果的正确性。推荐在加载前后执行行数校验和汇总指标比对。

四、ETL编排：Airflow和Prefect是主流的编排工具。DAG定义需要关注任务依赖、重试策略和超时设置。分支逻辑（BranchPythonOperator）实现动态路径选择，Sensor实现外部依赖等待。

五、数据质量：从完整性、唯一性、准确性、一致性、时效性、引用完整性六个维度建立质量检查体系。质量评分量化数据健康度，质量看板提供直观的可视化监控。

六、数据血缘：记录数据从源头到目标的完整生命周期，支撑影响分析和故障排查。元数据管理是数据治理的基础，应与ETL管道自动化集成。

十、进一步思考

数据管道与ETL工作流是一个随着数据规模和技术演进不断发展的领域。以下方向值得持续关注和实践：

实时化趋势：从批处理ETL向流处理ELT演进，使用Apache Flink、Kafka Streams实现秒级延迟的数据管道
DataOps实践：将CI/CD、版本控制、自动化测试引入数据管道管理，实现数据交付的敏捷化和自动化
AI辅助ETL：利用Claude Code等AI工具自动生成转换逻辑、异常检测规则和数据质量报告
数据网格：将数据按业务域分散治理，每个域拥有独立的数据管道，避免中心化ETL的性能瓶颈
成本优化：合理选择计算引擎（Pandas vs Spark vs Dask），在数据量和计算成本之间找到平衡
可观测性：构建完善的数据管道监控体系，包括延迟告警、数据倾斜检测、资源利用率追踪

"数据管道的终极目标不是处理数据，而是创造可信的、可追溯的、可理解的数据资产。Claude Code让数据工程师能专注于业务逻辑和架构设计，而不是被模板代码和调试细节所淹没。"