ETL精通：分布式调度监控与性能优化

┌─────────────────────────────────────────┐
│              Master Node                │
│  ┌──────────┐  ┌──────────┐            │
│  │ Scheduler │  │ Executor │            │
│  └─────┬────┘  └────┬─────┘            │
│        │             │                  │
│  ┌─────▼─────────────▼──────┐           │
│  │     Metadata DB          │           │
│  └──────────────────────────┘           │
└────────────────┬────────────────────────┘
                 │ 任务分发
    ┌────────────┼────────────┐
    ▼            ▼            ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Worker1 │ │ Worker2 │ │ Worker3 │
│ (Slave) │ │ (Slave) │ │ (Slave) │
└─────────┘ └─────────┘ └─────────┘

# 伪代码：对等模式任务分配示意
class PeerNode:
    def __init__(self, node_id, peers):
        self.node_id = node_id
        self.peers = peers  # 其他节点列表
        self.leader = self.elect_leader()
    
    def elect_leader(self):
        # Raft协议实现Leader选举
        pass
    
    def assign_task(self, task):
        # 一致性哈希分配任务
        target = consistent_hash(task.task_id, self.peers)
        if target.node_id == self.node_id:
            self.execute(task)
        else:
            self.forward(target, task)

数据源 → [Extract Service] → Kafka/TubeMQ → [Transform Service] → 
Kafka → [Load Service] → 目标存储
         ↑                       ↑                       ↑
    [Schema Registry]     [UDF Engine]           [DLQ Manager]

from concurrent.futures import ThreadPoolExecutor, as_completed

def extract_table(table_name):
    """从源库抽取一张表数据（I/O密集）"""
    return query_source(f"SELECT * FROM {table_name}")

def multi_thread_extract(tables: list, max_workers=8):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(extract_table, t): t for t in tables}
        results = {}
        for future in as_completed(futures):
            table = futures[future]
            try:
                results[table] = future.result()
            except Exception as e:
                log.error(f"Extract failed for {table}: {e}")
                raise
    return results

import multiprocessing as mp
from functools import partial

def transform_chunk(chunk, rules: dict):
    """对数据分片执行转换规则（CPU密集）"""
    return chunk.apply(rules)

def parallel_transform(df, rules: dict, num_workers=None):
    num_workers = num_workers or mp.cpu_count()
    chunks = np.array_split(df, num_workers)
    
    with mp.Pool(num_workers) as pool:
        results = pool.map(partial(transform_chunk, rules=rules), chunks)
    
    return pd.concat(results)

import asyncio
import aiohttp

async def fetch_api(session, endpoint):
    async with session.get(endpoint) as resp:
        return await resp.json()

async def async_extract(endpoints: list):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_api(session, ep) for ep in endpoints]
        return await asyncio.gather(*tasks, return_exceptions=True)

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.utils.task_group import TaskGroup

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

with DAG(
    dag_id='etl_order_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule_interval='0 2 * * *',  # 每天凌晨2点
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
    tags=['etl', 'order', 'production'],
) as dag:
    
    start = DummyOperator(task_id='start')
    
    with TaskGroup(group_id='extract_layer') as extract:
        extract_orders = PythonOperator(
            task_id='extract_orders',
            python_callable=extract_table,
            op_kwargs={'table': 'orders'},
            pool='extract_pool',
        )
        extract_users = PythonOperator(
            task_id='extract_users',
            python_callable=extract_table,
            op_kwargs={'table': 'users'},
            pool='extract_pool',
        )
    
    with TaskGroup(group_id='transform_layer') as transform:
        clean_orders = PythonOperator(
            task_id='clean_orders',
            python_callable=clean_data,
            sla=timedelta(minutes=30),
        )
        join_data = PythonOperator(task_id='join_data', python_callable=join_orders_users)
    
    with TaskGroup(group_id='load_layer') as load:
        load_dwd = PythonOperator(task_id='load_dwd', python_callable=load_to_dwd)
        load_dws = PythonOperator(task_id='load_dws', python_callable=load_to_dws)
    
    start >> extract >> transform >> load

# 1. 链式依赖
task1 >> task2 >> task3

# 2. 扇入（Fan-in）——多个前置任务完成后执行
[task_a, task_b, task_c] >> task_join

# 3. 扇出（Fan-out）——一个任务触发多个下游
task_source >> [task_x, task_y, task_z]

# 4. 条件分支
from airflow.operators.python import BranchPythonOperator

def decide_branch(**context):
    execution_date = context['execution_date']
    if execution_date.isoweekday() <= 5:
        return 'weekday_process'
    return 'weekend_process'

branch = BranchPythonOperator(
    task_id='branch_decision',
    python_callable=decide_branch,
)
branch >> [weekday_process, weekend_process]

# 5. 跨DAG依赖——使用ExternalTaskSensor
from airflow.sensors.external_task import ExternalTaskSensor

wait_for_upstream = ExternalTaskSensor(
    task_id='wait_for_dw_layer',
    external_dag_id='dw_daily_load',
    external_task_id='dwd_complete',
    timeout=3600,
    poke_interval=60,
)

# celery_app.py
from celery import Celery
from celery.signals import task_failure, task_success

celery_app = Celery(
    'etl_worker',
    broker='redis://redis-cluster:6379/0',
    backend='redis://redis-cluster:6379/1',
)

celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,          # 任务完成后才ACK
    worker_prefetch_multiplier=1, # 每次只取一个任务，防止倾斜
    task_reject_on_worker_lost=True,
)

# tasks.py
@celery_app.task(bind=True, max_retries=5, default_retry_delay=60)
def etl_sync_task(self, source_table, target_table):
    """分布式ETL任务"""
    try:
        data = extract_data(source_table)
        transformed = transform_data(data)
        load_data(transformed, target_table)
        return {'status': 'success', 'rows': len(transformed)}
    except ConnectionError as exc:
        # 网络异常——可重试
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)
    except Exception as exc:
        # 业务异常——记录死信
        send_to_dlq(source_table, target_table, exc)
        raise

# airflow.cfg 关键配置（CeleryExecutor模式）
[core]
executor = CeleryExecutor
parallelism = 32
dag_concurrency = 16

[celery]
worker_concurrency = 8
broker_url = redis://redis-cluster:6379/0
result_backend = redis://redis-cluster:6379/1
flower_url_prefix = /flower

[celery_kubernetes]
# 如果使用混合模式，可配置K8S相关参数
namespace = airflow
worker_container_image = airflow-worker:2.8.0

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Master-1   │    │  Master-2   │    │  Master-3   │
│  (Active)   │◄──►│ (Standby)   │◄──►│ (Standby)   │
└──────┬──────┘    └─────────────┘    └─────────────┘
       │ RPC
┌──────▼────────────────────────────────────┐
│            Zookeeper (选主+服务发现)          │
└──────┬────────────────────────────────────┘
       │
┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
│  Worker-1   │ │  Worker-2   │ │  Worker-3   │
│ (Queue-A)   │ │ (Queue-B)   │ │ (Queue-A)   │
└─────────────┘ └─────────────┘ └─────────────┘

# 通过DolphinScheduler REST API提交ETL任务
import requests
import json

def submit_ds_process(process_name, tasks, tenant='etl_user'):
    """向DolphinScheduler提交工作流"""
    api_url = "http://ds-master:12345/dolphinscheduler/projects/1/process-definition"
    
    # 构建工作流定义
    process_def = {
        "name": process_name,
        "description": "分布式ETL任务",
        "tenantCode": tenant,
        "globalParams": [],
        "tasks": tasks,
        "timeout": 7200,
    }
    
    resp = requests.post(api_url, json=process_def)
    return resp.json()

# SQL任务示例——无需编写Python代码
sql_task = {
    "type": "SQL",
    "id": "task_extract_orders",
    "name": "抽取订单数据",
    "params": {
        "type": "MYSQL",
        "datasource": 1,  # 数据源ID
        "sql": "SELECT * FROM orders WHERE dt = '${dt}'",
        "sqlType": "0",   # 查询
    },
    "runFlag": "NORMAL",
    "maxRetryTimes": 3,
    "retryInterval": 5,
}
shell_task = {
    "type": "SHELL",
    "id": "task_transform",
    "name": "数据转换",
    "params": {
        "resourceList": [],
        "localParams": [],
        "rawScript": "spark-submit --class TransformJob transform.jar --date ${dt}"
    },
}

def build_incremental_query(table, last_max_id, batch_size=100000):
    """基于主键的分页增量抽取"""
    return f"""
        SELECT * FROM {table}
        WHERE id > {last_max_id}
        ORDER BY id
        LIMIT {batch_size}
    """

def partition_by_date(base_date, num_partitions=8):
    """按日期范围分片"""
    from datetime import timedelta
    chunk_size = timedelta(hours=24 // num_partitions)
    for i in range(num_partitions):
        start = base_date + i * chunk_size
        end = start + chunk_size
        yield start, end

# 动态调整并行度
def calc_optimal_parallelism(total_records, row_size_bytes, worker_memory_mb):
    """根据数据量和内存计算最佳并行度"""
    per_worker_limit = worker_memory_mb * 0.6 * 1024 * 1024  # 60%可用内存
    records_per_worker = per_worker_limit // row_size_bytes
    parallelism = max(1, total_records // records_per_worker)
    # 限制最大并行度
    return min(parallelism, 32)

import gc
import psutil
import os

def memory_safe_transform(iterable, batch_size=10000):
    """流式处理——每次只保留一个batch在内存"""
    batch = []
    for i, row in enumerate(iterable, 1):
        batch.append(row)
        if i % batch_size == 0:
            yield process_batch(batch)
            batch.clear()
            if i % (batch_size * 10) == 0:
                # 每处理10万行强制GC
                gc.collect()
                # 检查内存水位
                mem = psutil.Process(os.getpid()).memory_percent()
                if mem > 80:
                    log.warning(f"Memory usage high: {mem:.1f}%")
    if batch:
        yield process_batch(batch)

# 倾斜检测与自动处理
class SkewDetector:
    def __init__(self, threshold_ratio=3.0):
        self.threshold_ratio = threshold_ratio  # 最大/最小比值阈值
    
    def detect(self, partition_sizes: dict):
        """检测是否存在数据倾斜"""
        sizes = list(partition_sizes.values())
        if len(sizes) < 2:
            return False
        max_size = max(sizes)
        min_size = min(sizes)
        return max_size / max(min_size, 1) > self.threshold_ratio
    
    def resolve(self, key, skewed_partition, num_splits=8):
        """对倾斜分区进行打散处理——加盐（Salting）"""
        def add_salt(row):
            salt = random.randint(0, num_splits - 1)
            return f"{row[key]}_{salt}"
        return add_salt

# 使用示例：在Spark中处理倾斜JOIN
"""
-- 大表加盐，小表膨胀，解决倾斜Join
WITH salted_orders AS (
    SELECT *, 
           CONCAT(order_id, '_', FLOOR(RAND() * 10)) AS salted_key
    FROM orders
),
expanded_users AS (
    SELECT *, 
           EXPLODE(ARRAY(0,1,2,3,4,5,6,7,8,9)) AS salt
    FROM users
)
SELECT /*+ BROADCAST(expanded_users) */ *
FROM salted_orders o
JOIN expanded_users u 
  ON o.salted_key = CONCAT(u.user_id, '_', u.salt)
"""

# 批量写入vs逐条写入
def batch_load(rows, target_table, batch_size=5000):
    """批量写入——减少IO次数"""
    conn = get_target_conn()
    cursor = conn.cursor()
    batch = []
    for row in rows:
        batch.append(row)
        if len(batch) >= batch_size:
            cursor.executemany(
                f"INSERT INTO {target_table} VALUES (%s, %s, %s)",
                batch
            )
            conn.commit()
            batch.clear()
    if batch:
        cursor.executemany(...)
        conn.commit()

                  ┌─────────┐
                  │ queued  │
                  └────┬────┘
                       ↓
                  ┌─────────┐
        ┌────────►│ running │◄────────┐
        │         └────┬────┘         │
        │              │              │
        │         ┌────▼────┐         │
        │    ┌────│ success │────┐    │
        │    │    └─────────┘    │    │
        │    │                   │    │
        ▼    ▼                   ▼    ▼
    ┌──────┐ ┌──────┐      ┌──────────┐
    │retry │ │skipped│      │ failed   │
    └──────┘ └──────┘      └────┬─────┘
                                │
                          ┌─────▼─────┐
                          │ dead_letter│
                          └───────────┘

# 自定义任务事件监听
from airflow.models import TaskInstance
from airflow.utils.state import State

def task_state_listener(context):
    """任务状态变更回调——发送到监控系统"""
    ti: TaskInstance = context['task_instance']
    dag_id = ti.dag_id
    task_id = ti.task_id
    state = ti.state
    duration = (ti.end_date - ti.start_date).total_seconds() if ti.end_date else 0
    
    metrics = {
        'dag_id': dag_id,
        'task_id': task_id,
        'state': state,
        'duration_seconds': duration,
        'execution_date': str(ti.execution_date),
        'hostname': ti.hostname,
        'try_number': ti.try_number,
    }
    
    # 推送到Prometheus / InfluxDB
    push_to_monitoring(metrics)
    
    # 失败告警
    if state == State.FAILED:
        send_alert(
            title=f"[ETL Failed] {dag_id}.{task_id}",
            message=f"执行次数: {ti.try_number}\n耗时: {duration}s",
            severity='critical',
        )

# 智能重试策略
class SmartRetryPolicy:
    """根据异常类型决定重试行为"""
    
    RETRYABLE_EXCEPTIONS = (
        ConnectionError,
        TimeoutError,
        DatabaseError,
    )
    
    FATAL_EXCEPTIONS = (
        DataValidationError,
        SchemaMismatchError,
        PermissionError,
    )
    
    @classmethod
    def should_retry(cls, exception, try_number, max_retries=3):
        if isinstance(exception, cls.FATAL_EXCEPTIONS):
            return False, "不可恢复异常，停止重试"
        if try_number >= max_retries:
            return False, f"已达最大重试次数({max_retries})"
        return True, f"第{try_number}次重试"
    
    @classmethod
    def calc_backoff(cls, try_number):
        """指数退避：1min, 2min, 4min, 8min..."""
        return min(2 ** (try_number - 1) * 60, 3600)  # 上限1小时

告警层级：
┌──────────────────────────────────────────────┐
│  P0（Critical）：链路中断、数据丢失          │
│  通知方式：电话 + 短信 + 即时消息 + 邮件     │
│  响应要求：15分钟                            │
├──────────────────────────────────────────────┤
│  P1（Error）：任务失败、数据延迟              │
│  通知方式：即时消息 + 邮件                   │
│  响应要求：1小时                             │
├──────────────────────────────────────────────┤
│  P2（Warning）：数据量波动、执行时间异常      │
│  通知方式：日报/周报汇总                     │
│  响应要求：24小时                            │
└──────────────────────────────────────────────┘

# 多通道告警分发实现
class AlertDispatcher:
    """告警分发器——按级别路由到不同通道"""
    
    def __init__(self):
        self.channels = {
            'p0': [SmsChannel(), PhoneChannel(), IMChannel(), EmailChannel()],
            'p1': [IMChannel(), EmailChannel()],
            'p2': [EmailChannel()],
        }
    
    def dispatch(self, alert: Alert):
        channels = self.channels.get(alert.severity, self.channels['p2'])
        results = []
        for channel in channels:
            try:
                channel.send(alert)
                results.append(True)
            except Exception as e:
                log.error(f"Alert channel {channel.name} failed: {e}")
                results.append(False)
        # 如果所有通道都失败，升级告警
        if not any(results):
            self.escalate(alert)
    
    def escalate(self, alert: Alert):
        """告警升级——通知值班经理"""
        escalation_channel = PhoneChannel()
        escalation_alert = Alert(
            title=f"[ESCALATION] {alert.title}",
            message=f"原始告警通道全部失败，需人工介入\n{alert.message}",
            severity='p0',
        )
        escalation_channel.send(escalation_alert)

# 集成企业微信/钉钉机器人
class IMChannel(AlertChannel):
    def send(self, alert: Alert):
        webhook_url = self.get_webhook_url(alert.severity)
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "content": (
                    f"## ⚠ ETL告警通知\n"
                    f"**级别:** {alert.severity}\n"
                    f"**任务:** {alert.dag_id}.{alert.task_id}\n"
                    f"**时间:** {alert.timestamp}\n"
                    f"**详情:** {alert.message}\n"
                )
            }
        }
        requests.post(webhook_url, json=payload, timeout=5)

# Prometheus指标暴露示例
from prometheus_client import Histogram, Counter, Gauge, start_http_server

etl_duration = Histogram(
    'etl_task_duration_seconds',
    'ETL任务执行耗时分布',
    ['dag_id', 'task_id'],
    buckets=(60, 300, 600, 1800, 3600, 7200),
)

etl_rows_processed = Counter(
    'etl_rows_processed_total',
    '处理数据行数',
    ['dag_id', 'task_id', 'table_name'],
)

etl_queue_backlog = Gauge(
    'etl_queue_backlog',
    '任务队列积压数',
    ['queue_name'],
)

airflow_etl/
├── dags/
│   ├── etl_order_pipeline.py      # 主DAG定义
│   └── utils/
│       ├── extractors.py          # 抽取模块
│       ├── transformers.py        # 转换模块
│       ├── loaders.py             # 加载模块
│       └── monitoring.py          # 监控埋点
├── plugins/
│   └── operators/
│       └── etl_operators.py       # 自定义Operator
├── config/
│   ├── airflow.cfg                # Airflow配置
│   └── connections.yaml           # 连接信息
└── requirements.txt

# plugins/operators/etl_operators.py
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from typing import Callable, Dict, Any
import time

class SmartETLOperator(BaseOperator):
    """
    智能ETL Operator —— 内置监控埋点、重试、性能统计
    """
    
    @apply_defaults
    def __init__(
        self,
        extract_fn: Callable,
        transform_fn: Callable,
        load_fn: Callable,
        source_config: Dict[str, Any],
        target_config: Dict[str, Any],
        batch_size: int = 10000,
        enable_metric: bool = True,
        *args, **kwargs
    ):
        super().__init__(*args, **kwargs)
        self.extract_fn = extract_fn
        self.transform_fn = transform_fn
        self.load_fn = load_fn
        self.source_config = source_config
        self.target_config = target_config
        self.batch_size = batch_size
        self.enable_metric = enable_metric
    
    def execute(self, context):
        dag_id = context['dag'].dag_id
        task_id = self.task_id
        
        log.info(f"Starting ETL: {dag_id}.{task_id}")
        start_time = time.time()
        total_rows = 0
        
        try:
            # 1. Extract——流式读取
            for batch_idx, raw_batch in enumerate(self.extract_fn(
                self.source_config, self.batch_size
            )):
                # 2. Transform
                transformed = self.transform_fn(raw_batch)
                
                # 3. Load
                self.load_fn(transformed, self.target_config)
                total_rows += len(transformed)
                
                if batch_idx % 10 == 0:
                    log.info(f"Processed {total_rows} rows...")
            
            # 上报指标
            duration = time.time() - start_time
            if self.enable_metric:
                report_etl_metrics(dag_id, task_id, total_rows, duration)
            
            log.info(f"ETL completed: {total_rows} rows in {duration:.2f}s")
            return {'rows': total_rows, 'duration': duration}
            
        except Exception as e:
            log.error(f"ETL failed: {e}")
            raise

# 启动Celery Worker（生产建议使用Supervisor或K8S管理）
celery -A etl_worker.celery_app worker \
    --loglevel=INFO \
    --concurrency=8 \
    --queues=etl_high,etl_normal \
    --hostname=worker1@%h \
    --max-tasks-per-child=1000 \
    --time-limit=7200 \
    --soft-time-limit=6600

                     ┌──────────────────┐
                     │   Nginx反向代理   │
                     └────────┬─────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
       ┌────────────┐ ┌────────────┐ ┌────────────┐
       │ Airflow    │ │ Airflow    │ │ Airflow    │
       │ Webserver  │ │ Scheduler  │ │ Flower     │
       └────────────┘ └─────┬──────┘ └────────────┘
                            │
              ┌─────────────┼─────────────┐
              ▼             ▼             ▼
       ┌──────────┐ ┌──────────┐ ┌──────────┐
       │ Celery   │ │ Celery   │ │ Celery   │
       │ Worker   │ │ Worker   │ │ Worker   │   ← 动态扩缩容
       └──────────┘ └──────────┘ └──────────┘
              │             │             │
              └─────────────┼─────────────┘
                            ▼
                     ┌──────────────┐
                     │    Redis     │
                     │  (Broker+   │
                     │   Backend)  │
                     └──────┬──────┘
                            ▼
                     ┌──────────────┐
                     │  PostgreSQL  │
                     │ (MetadataDB) │
                     └──────────────┘

     数据源： MySQL ──► Kafka ──► HDFS/Hive

# 数据校验示例
def validate_data_integrity(source_conn, target_conn, table):
    """基于行数和校验和的端到端校验"""
    source_count = source_conn.fetch_val(f"SELECT COUNT(*) FROM {table}")
    target_count = target_conn.fetch_val(f"SELECT COUNT(*) FROM {table}")
    
    source_checksum = source_conn.fetch_val(
        f"SELECT BIT_XOR(CAST(CRC32(CONCAT_WS('|', *)) AS UNSIGNED)) FROM {table}"
    )
    target_checksum = target_conn.fetch_val(
        f"SELECT BIT_XOR(CAST(CRC32(CONCAT_WS('|', *)) AS UNSIGNED)) FROM {table}"
    )
    
    assert source_count == target_count, f"行数不匹配: {source_count} vs {target_count}"
    assert source_checksum == target_checksum, f"校验和不匹配"
    return True

# 血缘元数据Schema（可存入Neo4j或Atlas）
lineage_record = {
    'source': {
        'type': 'mysql',
        'database': 'source_db',
        'table': 'orders',
        'columns': ['order_id', 'user_id', 'amount', 'created_at'],
    },
    'transform': [
        {'type': 'filter', 'description': '过滤取消订单'},
        {'type': 'join', 'with': 'users', 'keys': ['user_id']},
        {'type': 'aggregation', 'description': '按天聚合'},
    ],
    'target': {
        'type': 'hive',
        'database': 'dwd',
        'table': 'dwd_order_daily',
        'columns': ['dt', 'total_orders', 'total_amount'],
    },
    'execution': {
        'dag_id': 'etl_order_pipeline',
        'task_id': 'transform_and_load',
        'execution_time': '2024-06-15T02:30:00Z',
        'rows_written': 125000,
    },
}

# 数据质量规则引擎示例
class DataQualityEngine:
    """内置规则引擎，在ETL过程中实时校验数据质量"""
    
    RULES = {
        'not_null': lambda col, df: df[col].isnull().sum() == 0,
        'unique': lambda col, df: df[col].is_unique,
        'range': lambda col, df, mn, mx: df[col].between(mn, mx).all(),
        'regex': lambda col, df, pattern: df[col].str.match(pattern).all(),
        'referential': lambda col, df, ref_tbl, ref_col: ...,
    }
    
    def validate(self, df, checks: list):
        for check in checks:
            rule_type = check['type']
            rule_fn = self.RULES[rule_type]
            result = rule_fn(check['column'], df, **check.get('params', {}))
            if not result:
                raise DataQualityError(
                    f"质量规则未通过: {check}"
                )
                # 可选：将异常数据写入DLQ

# 元数据配置示例 (YAML)
tables:
  - name: dwd_order_daily
    owner: data_team
    sla: 08:00  # 每日8点前必须完成
    upstream:
      - source_db.orders
      - source_db.users
    downstream:
      - ads.order_report
    retention: 30  # 保留30天分区
    partition:
      column: dt
      granularity: DAY
    quality_checks:
      - type: not_null
        columns: [order_id, dt]
      - type: unique
        columns: [order_id]
      - type: row_count
        min: 1000
        max: 10000000