批量处理与调度策略

import time
import schedule
from datetime import datetime
from typing import Callable


def etl_extract():
    print(f"[{datetime.now()}] 开始抽取数据...")
    time.sleep(2)
    print(f"[{datetime.now()}] 抽取完成")


def etl_transform():
    print(f"[{datetime.now()}] 开始转换数据...")
    time.sleep(3)
    print(f"[{datetime.now()}] 转换完成")


def etl_load():
    print(f"[{datetime.now()}] 开始加载数据...")
    time.sleep(1)
    print(f"[{datetime.now()}] 加载完成")


def full_pipeline():
    """完整 ETL 流水线"""
    print("=" * 50)
    print(f"ETL 流水线开始: {datetime.now()}")
    etl_extract()
    etl_transform()
    etl_load()
    print(f"ETL 流水线结束: {datetime.now()}")
    print("=" * 50)


# 配置调度
schedule.every().day.at("03:00").do(full_pipeline)
schedule.every(10).minutes.do(etl_extract)

# 调度循环
if __name__ == "__main__":
    print("调度器已启动...")
    while True:
        schedule.run_pending()
        time.sleep(1)

import time
import functools
from datetime import datetime


def retry(max_attempts=3, delay=5):
    """重试装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    print(f"[{datetime.now()}] 第 {attempt} 次执行失败: {e}")
                    if attempt < max_attempts:
                        print(f"等待 {delay} 秒后重试...")
                        time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator


@retry(max_attempts=3, delay=5)
def fetch_from_api(url):
    """模拟从 API 抽取数据"""
    print(f"正在请求: {url}")
    # 模拟网络不稳定
    import random
    if random.random() < 0.5:
        raise ConnectionError("网络连接失败")
    return {"status": "ok", "data": [1, 2, 3]}


if __name__ == "__main__":
    result = fetch_from_api("https://api.example.com/orders")
    print(f"结果: {result}")

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator

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

dag = DAG(
    "daily_order_etl",
    default_args=default_args,
    description="每日订单 ETL 处理",
    schedule_interval="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["etl", "订单"],
)


def extract_orders(**context):
    """从业务库抽取订单"""
    # 实际的抽取逻辑
    execution_date = context["execution_date"]
    print(f"抽取 {execution_date} 的订单数据")
    return {"records_count": 100000}


def transform_orders(**context):
    """清洗和转换订单数据"""
    ti = context["ti"]
    extract_result = ti.xcom_pull(task_ids="extract_orders")
    print(f"转换 {extract_result['records_count']} 条订单记录")
    return {"status": "success"}


def load_orders(**context):
    """加载到数据仓库"""
    print("写入数据仓库")
    return {"loaded": True}


start = DummyOperator(task_id="start", dag=dag)

extract = PythonOperator(
    task_id="extract_orders",
    python_callable=extract_orders,
    provide_context=True,
    dag=dag,
)

transform = PythonOperator(
    task_id="transform_orders",
    python_callable=transform_orders,
    provide_context=True,
    dag=dag,
)

load = PythonOperator(
    task_id="load_orders",
    python_callable=load_orders,
    provide_context=True,
    dag=dag,
)

end = DummyOperator(task_id="end", dag=dag)

# 定义依赖关系
start >> extract >> transform >> load >> end

def etl_with_retry(task_func, max_retries=3, backoff_factor=2):
    """带指数退避的重试"""
    for attempt in range(max_retries):
        try:
            task_func()
            return
        except Exception as e:
            wait = backoff_factor ** attempt * 10
            print(f"第 {attempt + 1} 次失败，{wait} 秒后重试: {e}")
            if attempt < max_retries - 1:
                time.sleep(wait)
            else:
                print("已达最大重试次数，任务失败")


# 使用示例
def risky_extract():
    import random
    if random.random() < 0.7:
        raise TimeoutError("连接超时")
    print("抽取成功")


etl_with_retry(risky_extract)

def send_alert(task_name, error_message):
    """发送任务失败告警"""
    alert_message = f"""
    【ETL 任务失败】
    任务名称: {task_name}
    失败时间: {datetime.now()}
    错误信息: {error_message}
    影响: 下游报表将停滞
    """
    # 这里可以集成邮件、Slack、钉钉等通知渠道
    print(f"发送告警: {alert_message}")
    # requests.post(slack_webhook_url, json={"text": alert_message})

-- 使用 MERGE (UPSERT) 实现幂等加载
MERGE INTO dw.orders AS target
USING (
    SELECT order_id, amount, status, updated_at
    FROM staging.orders
    WHERE dt = CURRENT_DATE
) AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN
    UPDATE SET amount = source.amount, status = source.status,
              updated_at = source.updated_at
WHEN NOT MATCHED THEN
    INSERT (order_id, amount, status, updated_at)
    VALUES (source.order_id, source.amount, source.status, source.updated_at);

def incremental_etl(db_conn, watermark_table, source_table, target_table):
    """基于水印表的增量 ETL"""
    
    # 1. 读取上次的水印（处理到哪里的标记）
    cursor = db_conn.cursor()
    cursor.execute(f"SELECT last_processed_id FROM {watermark_table}")
    last_id = cursor.fetchone()[0]
    
    # 2. 抽取增量数据
    cursor.execute(f"""
        SELECT * FROM {source_table}
        WHERE id > {last_id}
        ORDER BY id
        LIMIT 10000
    """)
    batch = cursor.fetchall()
    
    while batch:
        # 3. 转换并加载这批数据
        load_to_target(batch, target_table)
        
        # 4. 更新水印
        max_id = batch[-1]["id"]
        cursor.execute(f"""
            UPDATE {watermark_table}
            SET last_processed_id = {max_id}, 
                updated_at = NOW()
        """)
        db_conn.commit()
        
        # 5. 处理下一批
        cursor.execute(f"""
            SELECT * FROM {source_table}
            WHERE id > {max_id}
            ORDER BY id
            LIMIT 10000
        """)
        batch = cursor.fetchall()
    
    print("增量 ETL 完成")