Python 数据工程与 ETL 管道实战

import pandas as pd
import numpy as np

# 模拟电商平台订单数据
orders = pd.DataFrame({
    'order_id': range(1, 11),
    'customer_id': ['A', 'A', 'B', 'B', 'B', 'C', 'C', 'D', 'D', 'D'],
    'amount': [100, 150, 200, 80, 120, 300, 50, 90, 110, 130],
    'category': ['elec', 'elec', 'book', 'book', 'elec', 'elec', 'book', 'elec', 'elec', 'book'],
    'order_date': pd.date_range('2024-01-01', periods=10, freq='D')
})

# 命名聚合：明确指定输出列名与计算方式
agg = orders.groupby('customer_id').agg(
    total_amount=('amount', 'sum'),
    avg_amount=('amount', 'mean'),
    order_count=('order_id', 'nunique'),
    max_order=('amount', 'max'),
    std_amount=('amount', 'std')
).round(2)

# 自定义聚合函数：计算每个客户是否产生复购
def repurchase_rate(x):
    return (x > 1).mean()

customer_stats = orders.groupby('customer_id').agg(
    orders=('order_id', 'count'),
    repurchase=('order_id', repurchase_rate)
)

# 模拟客户维度表，补充地域与等级信息
customers = pd.DataFrame({
    'customer_id': ['A', 'B', 'C', 'D', 'E'],
    'region': ['East', 'West', 'East', 'South', 'West'],
    'vip_level': [1, 2, 1, 3, 2]
})

# merge 等值连接：最常用
df = orders.merge(customers, on='customer_id', how='left')

# 多键连接：适用于联合主键场景
left = pd.DataFrame({'k1': ['A', 'B'], 'k2': [1, 2], 'v1': [10, 20]})
right = pd.DataFrame({'k1': ['A', 'B'], 'k2': [1, 3], 'v2': [100, 200]})
merged = pd.merge(left, right, on=['k1', 'k2'], how='outer')

# join：当两边 DataFrame 已按目标字段设为索引时使用
df1 = df.set_index('customer_id')
df2 = customers.set_index('customer_id')
joined = df1.join(df2, rsuffix='_customer')

# 透视表：用客户作为行、品类作为列观察销售矩阵
pivot = pd.pivot_table(
    orders,
    values='amount',
    index='customer_id',
    columns='category',
    aggfunc='sum',
    fill_value=0
)

# melt：宽表转长表，适合 tidy data 规范
long_format = pivot.reset_index().melt(
    id_vars='customer_id',
    var_name='category',
    value_name='amount'
)

# crosstab：快速生成交叉频数表
freq = pd.crosstab(orders['customer_id'], orders['category'], margins=True)

# 按客户计算累计消费金额
df_sorted = orders.sort_values(['customer_id', 'order_date'])
df_sorted['cumsum'] = df_sorted.groupby('customer_id')['amount'].cumsum()

# 滚动平均：7 日移动窗口，适合平滑短期波动
orders['ma_7d'] = orders.set_index('order_date')['amount'].rolling('7D').mean().values

# 分组排名：找出每个客户消费金额最高的订单排名
df_sorted['rank_in_customer'] = df_sorted.groupby('customer_id')['amount'].rank(method='dense', ascending=False)

# shift / diff：计算环比，识别增长速度
sales_daily = orders.groupby('order_date')['amount'].sum().reset_index()
sales_daily['prev_day'] = sales_daily['amount'].shift(1)
sales_daily['mom'] = sales_daily['amount'].pct_change()

import polars as pl

# 极速读取，列式格式 Parquet 为首选
df = pl.read_parquet('orders/*.parquet')

# 惰性求值：构建查询计划但暂不执行
lazy_df = (
    pl.scan_parquet('orders/*.parquet')
    .filter(pl.col('amount') > 100)
    .with_columns(
        (pl.col('amount') * 0.08).alias('tax'),
        pl.col('order_date').str.to_datetime('%Y-%m-%d').alias('dt')
    )
    .group_by('customer_id')
    .agg([
        pl.col('amount').sum().alias('total'),
        pl.col('amount').mean().round(2).alias('avg'),
        pl.col('order_id').count().alias('cnt')
    ])
    .sort('total', descending=True)
)

result = lazy_df.collect()  # 触发执行并物化结果

# 条件表达式：为金额分级
expr = (
    pl.when(pl.col('amount') >= 200)
    .then(pl.lit('high'))
    .when(pl.col('amount') >= 100)
    .then(pl.lit('medium'))
    .otherwise(pl.lit('low'))
    .alias('amount_tier')
)

df = df.with_columns(expr)

# 窗口函数：无需设置索引，直接在语义层面表达
df = df.with_columns(
    pl.col('amount').sum().over('customer_id').alias('customer_total'),
    pl.col('amount').shift(1).over('customer_id').alias('prev_amount')
)

# Streaming 模式处理超大数据（超出内存的文件）
result = (
    pl.scan_csv('huge_file.csv')
    .group_by('category')
    .agg(pl.col('amount').sum())
    .collect(streaming=True)  # 分块流式处理
)

import pandera as pa
from pandera import Column, Check, DataFrameSchema

# 定义订单数据契约：列名、类型、约束与严格模式
order_schema = DataFrameSchema({
    'order_id': Column(int, checks=Check.greater_than(0)),
    'customer_id': Column(str, checks=Check.str_length(min_value=1, max_value=10)),
    'amount': Column(float, checks=[
        Check.greater_than(0),
        Check.less_than_or_equal_to(100000)
    ]),
    'category': Column(str, checks=Check.isin(['elec', 'book', 'clothing', 'food'])),
    'order_date': Column('datetime64[ns]')
}, strict=True)  # strict=True 禁止出现未声明的列

# 执行校验，异常时抛 SchemaError
validated = order_schema.validate(orders)

# 装饰器模式：自动校验函数输出
check_output = pa.check_output(order_schema)

def clean_orders(raw_df):
    return raw_df.dropna(subset=['amount'])

import great_expectations as gx

context = gx.get_context()
source = context.data_sources.add_pandas('orders_source')
data_asset = source.add_dataframe_asset(name='orders_asset')

# 定义期望批次并执行校验
batch_definition = data_asset.add_batch_definition_whole_dataframe('batch_def')
batch = batch_definition.get_batch(batch_parameters={'dataframe': orders})

# 常见业务期望
results = []
results.append(batch.expect_column_values_to_not_be_null('customer_id'))
results.append(batch.expect_column_values_to_be_between('amount', min_value=0, max_value=1e6))
results.append(batch.expect_column_values_to_be_in_set('category', ['elec', 'book', 'clothing']))
results.append(batch.expect_column_pair_values_to_be_equal('order_date', 'order_date'))

for r in results:
    status = '通过' if r.success else '失败'
    print(f"{r.expectation_config.type}: {status}")

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.utils.dates import days_ago
from datetime import timedelta

import pandas as pd
import requests

def extract_orders(**context):
    """从 REST API 抽取订单数据并推送至 XCom"""
    resp = requests.get('https://api.example.com/orders', timeout=30)
    resp.raise_for_status()
    data = resp.json()['orders']
    df = pd.DataFrame(data)
    context['ti'].xcom_push(key='orders_df', value=df.to_json())
    return f"成功抽取 {len(df)} 条订单"

def transform_orders(**context):
    """从 XCom 拉取数据并执行清洗与业务转换"""
    ti = context['ti']
    json_data = ti.xcom_pull(task_ids='extract', key='orders_df')
    df = pd.read_json(json_data)

    df['amount'] = df['amount'].abs()
    df = df.dropna(subset=['customer_id', 'amount'])
    df['category'] = df['category'].str.lower().str.strip()
    df['order_date'] = pd.to_datetime(df['order_date'])

    ti.xcom_push(key='clean_df', value=df.to_json(date_format='iso'))
    return f"清洗后剩余 {len(df)} 条"

def load_to_warehouse(**context):
    """从 XCom 拉取清洗数据并写入数据仓库"""
    ti = context['ti']
    json_data = ti.xcom_pull(task_ids='transform', key='clean_df')
    df = pd.read_json(json_data)
    print(f"准备写入 {len(df)} 行到 warehouse")

with DAG(
    dag_id='etl_orders_pipeline',
    default_args={
        'owner': 'data-team',
        'retries': 2,
        'retry_delay': timedelta(minutes=5),
    },
    description='每日订单 ETL 管道',
    schedule_interval='@daily',
    start_date=days_ago(1),
    catchup=False,
    tags=['etl', 'orders'],
) as dag:

    extract = PythonOperator(task_id='extract', python_callable=extract_orders)

    transform = PythonOperator(task_id='transform', python_callable=transform_orders)

    create_staging = PostgresOperator(
        task_id='create_staging_table',
        postgres_conn_id='warehouse_conn',
        sql="""
            CREATE TABLE IF NOT EXISTS staging_orders (
                order_id INT PRIMARY KEY,
                customer_id VARCHAR(20),
                amount DECIMAL(12,2),
                category VARCHAR(50),
                order_date TIMESTAMP
            );
        """
    )

    load = PythonOperator(task_id='load', python_callable=load_to_warehouse)

    extract >> transform >> create_staging >> load

from dagster import asset, AssetIn, Definitions, schedule, define_asset_job, DefaultScheduleStatus
import pandas as pd
import requests

@asset(group_name="orders")
def raw_orders():
    """原始订单数据资产：从 API 拉取"""
    resp = requests.get('https://api.example.com/orders', timeout=30)
    resp.raise_for_status()
    return pd.DataFrame(resp.json()['orders'])

@asset(ins={"raw_orders": AssetIn()})
def cleaned_orders(raw_orders):
    """清洗后的订单资产：去空值、标准化、类型转换"""
    df = raw_orders.copy()
    df['amount'] = df['amount'].abs()
    df = df.dropna(subset=['customer_id', 'amount'])
    df['order_date'] = pd.to_datetime(df['order_date'])
    return df

@asset(ins={"cleaned_orders": AssetIn()})
def daily_revenue(cleaned_orders):
    """每日营收汇总资产"""
    return (
        cleaned_orders
        .groupby(cleaned_orders['order_date'].dt.date)
        .agg(total=('amount', 'sum'), count=('order_id', 'count'))
        .reset_index()
    )

# 定义作业与每日调度
orders_job = define_asset_job("orders_job", selection="*orders*")

daily_schedule = schedule(
    cron_schedule="0 2 * * *",
    target=orders_job,
    default_status=DefaultScheduleStatus.RUNNING,
)

defs = Definitions(
    assets=[raw_orders, cleaned_orders, daily_revenue],
    jobs=[orders_job],
    schedules=[daily_schedule],
)

from datetime import datetime, timedelta

# 上次同步时间存储在 Airflow Variable、Redis 或专用控制表中
last_sync = '2024-01-15 00:00:00'

def extract_incremental(conn, last_sync_time):
    query = """
        SELECT * FROM orders
        WHERE updated_at > %s
        ORDER BY updated_at ASC
    """
    return pd.read_sql(query, conn, params=(last_sync_time,))

# 动态滑动窗口：每小时抽取过去一小时的数据
start = (datetime.utcnow() - timedelta(hours=1)).strftime('%Y-%m-%d %H:%M:%S')

def extract_by_cursor(conn, last_id):
    """按自增 ID 分页抽取"""
    query = "SELECT * FROM orders WHERE id > %s ORDER BY id ASC LIMIT 10000"
    return pd.read_sql(query, conn, params=(last_id,))

from confluent_kafka import Consumer

consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'etl-consumer',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['db.orders'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    event = json.loads(msg.value().decode('utf-8'))
    # event['op'] 字段标识变更类型: c(reate), u(pdate), d(elete)
    handle_cdc_event(event)

def compute_quality_metrics(df):
    """计算数据质量评分卡"""
    total_rows = len(df)
    completeness = (1 - df.isnull().sum() / total_rows).mean()
    uniqueness = df.nunique() / total_rows

    # 数值字段离群率（IQR 方法）
    numeric = df.select_dtypes(include=[np.number])
    outlier_rates = {}
    for col in numeric.columns:
        q1, q3 = numeric[col].quantile([0.25, 0.75])
        iqr = q3 - q1
        mask = (numeric[col] < q1 - 1.5 * iqr) | (numeric[col] > q3 + 1.5 * iqr)
        outlier_rates[col] = mask.mean()

    return {
        'completeness': round(completeness, 4),
        'outlier_rates': {k: round(v, 4) for k, v in outlier_rates.items()},
        'row_count': total_rows
    }

from scipy import stats

# z-score 方法：标记超过 3 个标准差的记录
z_scores = np.abs(stats.zscore(df['amount']))
anomalies = df[z_scores > 3]

# IsolationForest：适合高维度或复杂分布
from sklearn.ensemble import IsolationForest

model = IsolationForest(contamination=0.05, random_state=42)
df['anomaly'] = model.fit_predict(df[['amount']].values)
# 输出值为 -1 表示异常，1 表示正常

# 伪代码：告警触发逻辑
if metrics['completeness'] < 0.98:
    alert_manager.send("【严重】订单表完整性低于 98%，请立即排查上游数据源")

if metrics['outlier_rates']['amount'] > 0.05:
    alert_manager.send("【警告】订单金额离群率高于 5%，可能存在单价录入错误")

import os
import logging
from datetime import datetime
import pandas as pd
import requests
import pandera as pa
from sqlalchemy import create_engine

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

# 配置从环境变量读取，支持容器化部署
API_BASE = os.getenv('API_BASE', 'https://api.example.com')
DB_URI = os.getenv('WAREHOUSE_URI', 'postgresql://user:pass@localhost:5432/warehouse')

# 1. 抽取：支持分页与超时控制
def extract_orders(from_date: str, to_date: str) -> pd.DataFrame:
    url = f"{API_BASE}/orders"
    params = {'from': from_date, 'to': to_date, 'per_page': 1000}
    all_data = []
    page = 1

    while True:
        params['page'] = page
        resp = requests.get(url, params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        all_data.extend(data['orders'])
        if not data.get('has_more', False):
            break
        page += 1

    logger.info(f"API 返回 {len(all_data)} 条订单，共 {page} 页")
    return pd.DataFrame(all_data)

# 2. 校验：schema 契约保障
def validate_data(df: pd.DataFrame) -> pd.DataFrame:
    ORDER_SCHEMA = pa.DataFrameSchema({
        'order_id': pa.Column(int, pa.Check.greater_than(0)),
        'customer_id': pa.Column(str, pa.Check.str_length(min_value=1)),
        'amount': pa.Column(float, pa.Check.greater_than(0)),
        'status': pa.Column(str, pa.Check.isin(['pending', 'paid', 'shipped', 'cancelled'])),
    })
    try:
        return ORDER_SCHEMA.validate(df)
    except pa.errors.SchemaError as e:
        logger.error(f"数据校验失败: {e}")
        raise

# 3. 转换：标准化与派生字段
def transform_data(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df['amount'] = df['amount'].round(2)
    df['status'] = df['status'].str.lower().str.strip()
    df['created_at'] = pd.to_datetime(df['created_at']).dt.tz_localize(None)
    df['etl_loaded_at'] = datetime.utcnow()
    return df

# 4. 加载：UPSERT 实现幂等写入
def load_to_warehouse(df: pd.DataFrame, table: str = 'staging_orders'):
    engine = create_engine(DB_URI)
    with engine.begin() as conn:
        tmp = f"{table}_tmp"
        df.to_sql(tmp, conn, if_exists='replace', index=False)
        conn.execute(f"""
            INSERT INTO {table}
            SELECT * FROM {tmp}
            ON CONFLICT (order_id) DO UPDATE SET
                customer_id = EXCLUDED.customer_id,
                amount = EXCLUDED.amount,
                status = EXCLUDED.status,
                etl_loaded_at = EXCLUDED.etl_loaded_at;
        """)
        conn.execute(f"DROP TABLE {tmp}")
    logger.info(f"写入 {table} 完成: {len(df)} 行")

# 5. 主控流程
def run_etl(from_date: str, to_date: str):
    raw = extract_orders(from_date, to_date)
    validated = validate_data(raw)
    transformed = transform_data(validated)
    load_to_warehouse(transformed)
    metrics = {
        'rows_extracted': len(raw),
        'rows_loaded': len(transformed),
        'quality_score': 1.0 if len(raw) == len(transformed) else len(transformed) / len(raw)
    }
    logger.info(f"ETL 完成: {metrics}")
    return metrics

if __name__ == '__main__':
    today = datetime.utcnow().strftime('%Y-%m-%d')
    run_etl(today, today)