[PostgreSQL] PostgreSQL+Python：构建实时模型训练数据管道

# 编辑PostgreSQL配置文件
sudo nano /etc/postgresql/15/main/postgresql.conf

# 关键配置项修改如下
wal_level = logical           # 启用逻辑复制
max_replication_slots = 10    # 最大复制槽数量
max_wal_senders = 10          # WAL发送进程数

# 允许复制连接
sudo nano /etc/postgresql/15/main/pg_hba.conf

# 添加以下行
host replication debezium 0.0.0.0/0 md5  # 允许debezium用户复制
host all all 0.0.0.0/0 md5               # 允许所有IP连接（测试环境）

-- 创建用于CDC的数据库用户
CREATE USER debezium WITH PASSWORD 'dbz_p@ssw0rd' REPLICATION LOGIN;

-- 授予对所有业务表的SELECT权限
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;

-- 创建复制槽（可通过Debezium自动创建）
SELECT * FROM pg_create_logical_replication_slot('debezium_slot', 'pgoutput');

# 创建项目目录结构
mkdir pg_ml_pipeline && cd pg_ml_pipeline
mkdir -p {config,src/{consumers,features,models},tests,scripts}

# 创建Python虚拟环境
python3.10 -m venv venv
source venv/bin/activate

# 安装核心依赖（带版本锁定）
pip install faust-streaming==0.10.4 psycopg2-binary==2.9.7 \
    confluent_kafka==2.2.0 mlflow==2.7.1 pandas==2.0.3 \
    numpy==1.24.3 scikit-learn==1.3.0 redis==5.0.1

# 导出依赖清单
pip freeze > requirements.txt

version: '3.8'
services:
  postgres:
    image: postgres:15-alpine
    container_name: pg_ml_pipeline
    environment:
      POSTGRES_USER: ml_user
      POSTGRES_PASSWORD: ml_password
      POSTGRES_DB: ecommerce_db
    ports:
      - "5432:5432"
    volumes:
      - pg_data:/var/lib/postgresql/data
      - ./config/postgresql.conf:/etc/postgresql/postgresql.conf
    command: postgres -c config_file=/etc/postgresql/postgresql.conf

  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  debezium:
    image: debezium/connect:2.5
    container_name: debezium
    depends_on:
      - kafka
      - postgres
    ports:
      - "8083:8083"
    environment:
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: debezium_config
      OFFSET_STORAGE_TOPIC: debezium_offsets
      BOOTSTRAP_SERVERS: kafka:9092

volumes:
  pg_data:

# 后台启动所有服务
docker-compose up -d

# 验证PostgreSQL连接
docker exec -it pg_ml_pipeline psql -U ml_user -d ecommerce_db -c "SELECT version();"

# 验证Kafka
docker exec -it kafka kafka-topics --list --bootstrap-server localhost:9092

# 验证Debezium
curl -H "Accept:application/json" localhost:8083/

-- 用户行为表（核心事实表）
CREATE TABLE user_behaviors (
    behavior_id BIGSERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    behavior_type VARCHAR(20) NOT NULL, -- 'view', 'cart', 'purchase'
    behavior_timestamp TIMESTAMP DEFAULT NOW(),
    session_id VARCHAR(100),
    device_type VARCHAR(20)
);

-- 用户画像表（维度表）
CREATE TABLE user_profiles (
    user_id INTEGER PRIMARY KEY,
    age_group VARCHAR(10),
    membership_level VARCHAR(20),
    register_date DATE
);

-- 商品信息表（维度表）
CREATE TABLE products (
    product_id INTEGER PRIMARY KEY,
    category_id INTEGER,
    price DECIMAL(10,2),
    brand VARCHAR(100),
    stock_quantity INTEGER
);

curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  localhost:8083/connectors/ -d '{
  "name": "ecommerce-user-behaviors-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz_p@ssw0rd",
    "database.dbname": "ecommerce_db",
    "database.server.name": "ecommerce_server",
    "table.include.list": "public.user_behaviors",
    "plugin.name": "pgoutput",
    "topic.prefix": "cdc.ecommerce",
    "tombstones.on.delete": "false",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "true",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}'

{
  "behavior_id": 12345,
  "user_id": 67890,
  "product_id": 54321,
  "behavior_type": "cart",
  "behavior_timestamp": "2025-12-01T14:30:22Z",
  "session_id": "sess_abc123",
  "device_type": "mobile"
}

# 消费时需要注意的字段类型转换
timestamp_mapping = {
    # PostgreSQL的TIMESTAMP带时区信息，需要转换
    "behavior_timestamp": lambda x: datetime.fromisoformat(x.replace('Z', '+00:00'))
}

# 处理可能的null值
null_fields = ["session_id", "device_type"]  # 这些字段可能为空

src/
├── __init__.py
├── consumers/
│   ├── __init__.py
│   ├── behavior_consumer.py  # 主消费逻辑
│   ├── feature_calculator.py # 特征计算
│   └── sink.py               # 数据写入
├── features/
│   ├── __init__.py
│   ├── realtime_features.py  # 实时特征定义
│   └── windowed_features.py  # 窗口特征
└── config/
    └── settings.py           # 配置管理

import faust
import json
from datetime import datetime, timedelta
from typing import Dict, Any
import psycopg2
from psycopg2.extras import RealDictCursor
import redis

# Faust应用初始化
app = faust.App(
    'behavior_pipeline',
    broker='kafka://localhost:9092',
    value_serializer='json',
    consumer_auto_offset_reset='earliest',
    # 消费者组配置
    group_id='ml_pipeline_group',
    # 故障恢复配置
    processing_guarantee='at_least_once',
    # 性能调优
    topic_partitions=6,  # 根据Kafka分区数调整
    fetch_max_wait_ms=100,
)

# 定义Topic
behavior_topic = app.topic('cdc.ecommerce.public.user_behaviors', 
                          value_type=Dict[str, Any])

# 初始化连接池
class ConnectionPool:
    """管理数据库和Redis连接"""
    def __init__(self, pg_config: dict, redis_config: dict):
        self.pg_config = pg_config
        self.redis_config = redis_config
        self._pg_pool = None
        self._redis_client = None
    
    def get_pg_conn(self):
        """获取PostgreSQL连接"""
        if not self._pg_pool:
            self._pg_pool = psycopg2.pool.ThreadedConnectionPool(
                minconn=5,
                maxconn=20,
                **self.pg_config
            )
        return self._pg_pool.getconn()
    
    def get_redis(self):
        """获取Redis客户端"""
        if not self._redis_client:
            self._redis_client = redis.Redis(**self.redis_config, decode_responses=True)
        return self._redis_client
    
    def release_pg_conn(self, conn):
        """归还连接"""
        self._pg_pool.putconn(conn)

# 初始化全局连接池
pool = ConnectionPool(
    pg_config={
        'host': 'localhost',
        'database': 'ecommerce_db',
        'user': 'ml_user',
        'password': 'ml_password'
    },
    redis_config={
        'host': 'localhost',
        'port': 6379,
        'db': 0
    }
)

@app.agent(behavior_topic, concurrency=6)  # 6个并发worker
async def process_behavior(stream):
    """
    处理用户行为事件的主函数
    
    实现要点：
    1. 批量处理：每100条或每50ms刷新一次
    2. 并行计算：聚合、窗口、用户画像查询并行执行
    3. 异常隔离：单条记录失败不影响批次
    """
    async for batch in stream.take(100, within=timedelta(milliseconds=50)):
        # I. 数据预处理批次
        valid_events = []
        for event in batch:
            try:
                cleaned = await validate_and_clean(event)
                valid_events.append(cleaned)
            except Exception as e:
                await handle_invalid_event(event, str(e))
                continue
        
        if not valid_events:
            continue
        
        # II. 并行特征计算
        # 创建异步任务
        tasks = []
        for event in valid_events:
            tasks.append(
                calculate_features(
                    event,
                    pool.get_pg_conn(),
                    pool.get_redis()
                )
            )
        
        # 等待所有特征计算完成
        features_batch = await asyncio.gather(*tasks, return_exceptions=True)
        
        # III. 过滤失败任务
        successful_features = [f for f in features_batch if not isinstance(f, Exception)]
        
        # IV. 批量写入
        if successful_features:
            await write_features_batch(successful_features)
        
        # 释放连接
        for event in valid_events:
            pool.release_pg_conn(event['pg_conn'])

async def validate_and_clean(event: dict) -> dict:
    """
    数据清洗和验证
    
    关键转换：
    1. 时间戳标准化为UTC
    2. 枚举值小写化
    3. 空值处理
    """
    # 时间戳转换
    ts_str = event.get('behavior_timestamp')
    if ts_str:
        event['behavior_timestamp'] = datetime.fromisoformat(
            ts_str.replace('Z', '+00:00')
        )
    
    # 枚举值标准化
    behavior_type = event.get('behavior_type', '').lower()
    if behavior_type not in ['view', 'cart', 'purchase']:
        event['behavior_type'] = 'unknown'
    else:
        event['behavior_type'] = behavior_type
    
    # 设备类型默认处理
    event['device_type'] = event.get('device_type', 'unknown')
    
    return event

async def handle_invalid_event(event: dict, error: str):
    """处理脏数据"""
    conn = pool.get_pg_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO dead_letter_queue 
                (event_data, error_message, created_at)
                VALUES (%s, %s, NOW())
            """, (json.dumps(event), error))
        conn.commit()
    finally:
        pool.release_pg_conn(conn)

async def write_features_batch(features: list):
    """批量写入特征到PostgreSQL"""
    conn = pool.get_pg_conn()
    try:
        with conn.cursor() as cur:
            # 使用executemany提高性能
            execute_values(
                cur,
                """
                INSERT INTO user_features_realtime 
                (user_id, feature_name, feature_value, updated_at)
                VALUES %s
                ON CONFLICT (user_id, feature_name) 
                DO UPDATE SET 
                    feature_value = EXCLUDED.feature_value,
                    updated_at = EXCLUDED.updated_at
                """,
                features,
                template='(%(user_id)s, %(feature_name)s, %(feature_value)s, NOW())'
            )
        conn.commit()
    except Exception as e:
        conn.rollback()
        # 失败时降级到单条插入
        for feature in features:
            try:
                with conn.cursor() as cur:
                    cur.execute("""
                        INSERT INTO user_features_realtime VALUES (%s, %s, %s, NOW())
                        ON CONFLICT (user_id, feature_name) 
                        DO UPDATE SET feature_value = %s, updated_at = NOW()
                    """, (feature['user_id'], feature['feature_name'], 
                         feature['feature_value'], feature['feature_value']))
                conn.commit()
            except Exception as e2:
                conn.rollback()
                print(f"单条写入失败: {e2}")
    finally:
        pool.release_pg_conn(conn)

if __name__ == '__main__':
    # 启动Faust worker
    # faust -A behavior_consumer worker -l info --web-port 6066
    app.main()

import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Any

async def calculate_features(event: dict, pg_conn, redis_client) -> List[Dict]:
    """
    为单个事件计算所有实时特征
    
    特征类型：
    I. 计数特征：最近N次行为统计
    II. 时间特征：距上次行为时间间隔
    III. 比率特征：转化率计算
    IV. 画像特征：用户标签JOIN
    """
    user_id = event['user_id']
    product_id = event['product_id']
    behavior_type = event['behavior_type']
    event_time = event['behavior_timestamp']
    
    # 并行计算不同特征
    tasks = [
        calculate_behavior_count(user_id, behavior_type, redis_client),
        calculate_time_since_last(user_id, behavior_type, redis_client, event_time),
        calculate_user_profile_features(user_id, pg_conn),
        calculate_product_features(product_id, pg_conn)
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    features = []
    # I. 行为计数特征
    if not isinstance(results[0], Exception):
        features.append({
            'user_id': user_id,
            'feature_name': f'count_{behavior_type}_last_1h',
            'feature_value': results[0]['last_1h']
        })
        features.append({
            'user_id': user_id,
            'feature_name': f'count_{behavior_type}_last_24h',
            'feature_value': results[0]['last_24h']
        })
    
    # II. 时间间隔特征
    if not isinstance(results[1], Exception) and results[1] is not None:
        features.append({
            'user_id': user_id,
            'feature_name': f'time_since_last_{behavior_type}',
            'feature_value': results[1].total_seconds()
        })
    
    # III. 用户画像特征
    if not isinstance(results[2], Exception):
        features.extend(results[2])
    
    # IV. 商品特征
    if not isinstance(results[3], Exception):
        features.append({
            'user_id': user_id,
            'feature_name': 'product_price',
            'feature_value': results[3]['price']
        })
        features.append({
            'user_id': user_id,
            'feature_name': 'product_category',
            'feature_value': results[3]['category_id']
        })
    
    return features

async def calculate_behavior_count(user_id: int, behavior_type: str, 
                                  redis_client) -> Dict[str, int]:
    """
    计算用户最近行为次数
    
    实现策略：
    1. 使用Redis HyperLogLog进行去重计数
    2. 使用Sorted Set存储时间序列，自动过期
    """
    key_prefix = f"behavior:{user_id}:{behavior_type}"
    
    # 获取当前时间戳
    now = datetime.utcnow()
    now_ts = int(now.timestamp())
    
    # 1小时窗口
    hour_ago_ts = now_ts - 3600
    hour_key = f"{key_prefix}:hourly"
    
    # 添加当前事件到Sorted Set
    redis_client.zadd(hour_key, {f"event:{now_ts}": now_ts})
    
    # 清理过期数据
    redis_client.zremrangebyscore(hour_key, 0, hour_ago_ts)
    
    # 获取窗口内数量
    count_1h = redis_client.zcount(hour_key, hour_ago_ts, now_ts)
    
    # 24小时窗口（使用不同的key策略避免大key）
    day_key = f"{key_prefix}:daily:{now.strftime('%Y%m%d')}"
    redis_client.pfadd(day_key, f"{now_ts}")
    
    # 获取前一天数据用于计算
    yesterday = now - timedelta(days=1)
    yesterday_key = f"{key_prefix}:daily:{yesterday.strftime('%Y%m%d')}"
    
    # 合并两天数据
    pipeline = redis_client.pipeline()
    pipeline.pfcount(day_key)
    pipeline.pfcount(yesterday_key)
    counts = pipeline.execute()
    
    # 使用指数衰减估算24小时
    count_24h = counts[0] + int(counts[1] * 0.5)
    
    return {'last_1h': count_1h, 'last_24h': count_24h}

async def calculate_time_since_last(user_id: int, behavior_type: str,
                                   redis_client, current_time: datetime) -> timedelta:
    """
    计算距上次同类型行为的时间间隔
    
    状态管理：
    1. 使用Redis String存储最后一次时间戳
    2. 设置过期时间避免内存泄漏
    """
    key = f"last_time:{user_id}:{behavior_type}"
    
    # 获取上次时间（Redis返回的是字符串）
    last_ts_str = redis_client.get(key)
    
    # 更新为当前时间（设置1小时过期）
    current_ts = int(current_time.timestamp())
    redis_client.setex(key, 3600, current_ts)
    
    if last_ts_str:
        last_ts = int(last_ts_str)
        return current_time - datetime.fromtimestamp(last_ts, tz=current_time.tzinfo)
    
    return None

async def calculate_user_profile_features(user_id: int, pg_conn) -> List[Dict]:
    """查询用户画像特征"""
    with pg_conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute("""
            SELECT 
                age_group,
                membership_level,
                EXTRACT(DAYS FROM NOW() - register_date) as days_since_register
            FROM user_profiles
            WHERE user_id = %s
        """, (user_id,))
        
        profile = cur.fetchone()
        if not profile:
            return []
        
        return [
            {
                'user_id': user_id,
                'feature_name': 'age_group',
                'feature_value': hash(profile['age_group']) % 1000  # 类别哈希
            },
            {
                'user_id': user_id,
                'feature_name': 'membership_level',
                'feature_value': {'bronze': 1, 'silver': 2, 'gold': 3}.get(profile['membership_level'], 0)
            },
            {
                'user_id': user_id,
                'feature_name': 'user_lifetime_days',
                'feature_value': profile['days_since_register']
            }
        ]

async def calculate_product_features(product_id: int, pg_conn) -> Dict:
    """查询商品静态特征"""
    with pg_conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute("""
            SELECT category_id, price, brand
            FROM products
            WHERE product_id = %s
        """, (product_id,))
        
        product = cur.fetchone()
        if not product:
            return {'price': 0.0, 'category_id': -1, 'brand': 'unknown'}
        
        return {
            'price': float(product['price']),
            'category_id': product['category_id']
        }

-- 主表（实时特征）
CREATE TABLE user_features_realtime (
    user_id INTEGER NOT NULL,
    feature_name VARCHAR(100) NOT NULL,
    feature_value DOUBLE PRECISION,
    updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (user_id, feature_name)
) PARTITION BY RANGE (user_id);

-- 创建256个分区，平衡查询和写入
CREATE TABLE user_features_realtime_0 PARTITION OF user_features_realtime
    FOR VALUES FROM (0) TO (1000000);
CREATE TABLE user_features_realtime_1 PARTITION OF user_features_realtime
    FOR VALUES FROM (1000000) TO (2000000);
-- ... 依此类推创建更多分区

-- 索引优化
CREATE INDEX idx_features_updated ON user_features_realtime (updated_at);
CREATE INDEX idx_features_name_value ON user_features_realtime (feature_name, feature_value);

-- 历史特征表（按天分区，用于训练）
CREATE TABLE user_features historical (
    user_id INTEGER NOT NULL,
    feature_name VARCHAR(100) NOT NULL,
    feature_value DOUBLE PRECISION,
    date DATE NOT NULL,
    PRIMARY KEY (user_id, feature_name, date)
) PARTITION BY RANGE (date);

CREATE TABLE user_features_hist_202501 PARTITION OF user_features_historical
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

-- 死信队列
CREATE TABLE dead_letter_queue (
    id SERIAL PRIMARY KEY,
    event_data JSONB,
    error_message TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    retry_count INTEGER DEFAULT 0
) PARTITION BY RANGE (created_at);

# features/feature_store.py
import mlflow
from mlflow.tracking import MlflowClient
from datetime import datetime

class FeatureStore:
    """封装MLflow特征存储接口"""
    
    def __init__(self, tracking_uri: str):
        mlflow.set_tracking_uri(tracking_uri)
        self.client = MlflowClient()
    
    def log_feature_set(self, features: list, version: str):
        """
        记录特征集版本
        
        参数说明：
        - features: 特征定义列表
        - version: 语义化版本如v1.2.0
        """
        # 创建Feature Set
        feature_set = mlflow.entities.FeatureSpec(
            name="user_realtime_features",
            version=version,
            features=[
                mlflow.entities.Feature(
                    name=f['name'],
                    dtype=f['dtype'],
                    description=f['description']
                ) for f in features
            ]
        )
        
        # 注册到MLflow
        self.client.log_feature_spec(feature_set)
        
        # 打标签
        self.client.set_tag(
            "feature_set.created_at",
            datetime.utcnow().isoformat()
        )
    
    def get_feature_set(self, version: str = None) -> dict:
        """获取指定版本的特征定义"""
        # MLflow Feature Store API调用
        # 实现略...
        pass

# dags/feature_backfill.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml_pipeline',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'email_on_failure': True,
    'email': ['ml-team@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'daily_feature_backfill',
    default_args=default_args,
    description='每日补全历史特征',
    schedule_interval='0 2 * * *',  # 每天凌晨2点
    catchup=True,
    max_active_runs=1
) as dag:
    
    def backfill_features(ds, **kwargs):
        """补填T-1日的特征"""
        from src.features.windowed_features import calculate_daily_features
        
        target_date = datetime.strptime(ds, '%Y-%m-%d').date()
        calculate_daily_features(
            date=target_date,
            pg_config={
                'host': 'postgres',
                'database': 'ecommerce_db',
                'user': 'ml_user',
                'password': 'ml_password'
            }
        )
    
    backfill_task = PythonOperator(
        task_id='backfill_user_features',
        python_callable=backfill_features,
        execution_timeout=timedelta(hours=4)
    )

# models/training_pipeline.py
import mlflow
import mlflow.sklearn
import pandas as pd
from sqlalchemy import create_engine
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score, precision_recall_curve
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TrainingPipeline:
    """自动化训练流水线"""
    
    def __init__(self, pg_uri: str, experiment_name: str):
        self.engine = create_engine(pg_uri)
        mlflow.set_experiment(experiment_name)
        mlflow.set_tracking_uri("postgresql://ml_user:ml_password@postgres:5432/mlflowdb")
    
    def load_training_data(self, train_date: str, feature_version: str) -> pd.DataFrame:
        """
        从PostgreSQL加载训练数据
        
        SQL优化要点：
        1. 使用分区裁剪只扫描必要分区
        2. LEFT JOIN避免维度缺失导致数据丢失
        3. 窗口函数避免自连接
        """
        query = f"""
            WITH labeled_data AS (
                -- 创建标签：未来24小时内是否购买
                SELECT 
                    b.user_id,
                    b.behavior_timestamp as event_time,
                    CASE WHEN EXISTS (
                        SELECT 1 FROM user_behaviors b2
                        WHERE b2.user_id = b.user_id
                          AND b2.behavior_type = 'purchase'
                          AND b2.behavior_timestamp > b.behavior_timestamp
                          AND b2.behavior_timestamp <= b.behavior_timestamp + INTERVAL '24 hours'
                    ) THEN 1 ELSE 0 END as will_purchase
                FROM user_behaviors b
                WHERE b.behavior_timestamp::date = '{train_date}'
                  AND b.behavior_type = 'view'  -- 只预测浏览行为
            )
            SELECT 
                ld.user_id,
                ld.event_time,
                ld.will_purchase as label,
                -- 基础特征
                f1.feature_value as count_view_last_1h,
                f2.feature_value as count_cart_last_24h,
                f3.feature_value as time_since_last_view,
                f4.feature_value as membership_level,
                f5.feature_value as user_lifetime_days,
                f6.feature_value as product_price,
                -- 衍生特征
                (f6.feature_value * f4.feature_value) as price_x_membership,
                CASE WHEN f3.feature_value < 300 THEN 1 ELSE 0 END as is_frequent
            FROM labeled_data ld
            -- 特征JOIN（使用LATERAL优化）
            LEFT JOIN LATERAL (
                SELECT feature_value 
                FROM user_features_historical
                WHERE user_id = ld.user_id 
                  AND feature_name = 'count_view_last_1h'
                  AND date = '{train_date}'
                LIMIT 1
            ) f1 ON true
            -- 其他特征类似...
            WHERE ld.will_purchase IN (0,1)  -- 排除NULL标签
        """
        
        logger.info(f"执行查询: {query[:200]}...")
        df = pd.read_sql(query, self.engine)
        logger.info(f"加载数据: {df.shape[0]}行, {df.shape[1]}列")
        
        return df
    
    def train_model(self, df: pd.DataFrame, hyperparams: dict) -> dict:
        """
        训练随机森林模型
        
        实现要点：
        I. 数据分割：按用户ID分组，避免数据泄露
        II. 类别编码：对分类特征进行目标编码
        III. 模型训练：记录所有超参数
        IV. 评估：多维度指标计算
        """
        # 按用户ID分组分割，避免同一用户出现在训练集和测试集
        user_ids = df['user_id'].unique()
        train_users, test_users = train_test_split(
            user_ids, test_size=0.2, random_state=42
        )
        
        train_df = df[df['user_id'].isin(train_users)]
        test_df = df[df['user_id'].isin(test_users)]
        
        # 特征和标签
        feature_cols = [c for c in df.columns if c not in ['user_id', 'event_time', 'label']]
        X_train = train_df[feature_cols]
        y_train = train_df['label']
        X_test = test_df[feature_cols]
        y_test = test_df['label']
        
        # 训练
        with mlflow.start_run():
            # 记录参数
            mlflow.log_params(hyperparams)
            
            # 记录特征列表
            mlflow.log_dict({'features': feature_cols}, 'features.json')
            
            # 训练模型
            model = RandomForestClassifier(**hyperparams, random_state=42, n_jobs=-1)
            model.fit(X_train, y_train)
            
            # 预测
            y_pred_proba = model.predict_proba(X_test)[:, 1]
            y_pred = model.predict(X_test)
            
            # 评估指标
            auc = roc_auc_score(y_test, y_pred_proba)
            precision, recall, _ = precision_recall_curve(y_test, y_pred_proba)
            
            # 记录指标
            mlflow.log_metrics({
                'auc': auc,
                'precision_at_50': precision[recall >= 0.5][0],
                'avg_precision': precision.mean()
            })
            
            # 记录模型
            mlflow.sklearn.log_model(
                model, 
                'model',
                registered_model_name='user_purchase_predictor'
            )
            
            # 记录特征重要性
            importance_df = pd.DataFrame({
                'feature': feature_cols,
                'importance': model.feature_importances_
            }).sort_values('importance', ascending=False)
            
            mlflow.log_table(importance_df, 'feature_importance.json')
            
            logger.info(f"模型训练完成: AUC={auc:.4f}")
            
            return {
                'model_uri': f"runs:/{mlflow.active_run().info.run_id}/model",
                'metrics': {'auc': auc}
            }

if __name__ == '__main__':
    pipeline = TrainingPipeline(
        pg_uri="postgresql://ml_user:ml_password@localhost:5432/ecommerce_db",
        experiment_name="user_purchase_prediction"
    )
    
    # 训练最近7天的数据
    for i in range(1, 8):
        train_date = (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d')
        
        try:
            df = pipeline.load_training_data(train_date, feature_version='v1.0')
            
            if df.empty or df['label'].nunique() < 2:
                print(f"{train_date} 数据不足，跳过")
                continue
            
            result = pipeline.train_model(df, hyperparams={
                'n_estimators': 200,
                'max_depth': 8,
                'min_samples_split': 100,
                'min_samples_leaf': 50
            })
            
            print(f"{train_date}训练成功: {result}")
            
        except Exception as e:
            print(f"{train_date}训练失败: {e}")
            continue

# 在Faust consumer中嵌入触发逻辑
async def check_training_trigger(features_batch):
    """检查是否触发模型训练"""
    global event_counter
    
    event_counter += len(features_batch)
    
    # 每10万条事件触发一次增量训练
    if event_counter >= 100000:
        event_counter = 0
        
        # 异步触发Airflow DAG
        import requests
        requests.post(
            'http://airflow:8080/api/v1/dags/daily_feature_backfill/dagRuns',
            json={"conf": {"incremental": True}},
            auth=('admin', 'airflow_password')
        )

# 使用Python slim基础镜像
FROM python:3.10-slim

# 设置时区和环境变量
ENV TZ=Asia/Shanghai \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=1

# 安装系统依赖（包括PostgreSQL客户端和编译工具）
RUN apt-get update && apt-get install -y \
    gcc \
    postgresql-client \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# 创建非root用户
RUN groupadd -r mluser && useradd -r -g mluser mluser

# 设置工作目录
WORKDIR /app

# 先复制依赖文件，利用缓存
COPY requirements.txt .
RUN pip install --user --no-warn-script-location -r requirements.txt

# 复制应用代码
COPY src/ ./src/
COPY config/ ./config/

# 修改权限
RUN chown -R mluser:mluser /app

# 切换到非root用户
USER mluser

# 暴露监控端口
EXPOSE 6066

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s \
  CMD pg_isready -h postgres -p 5432 || exit 1

# 启动命令
CMD ["faust", "-A", "src.consumers.behavior_consumer", "worker", \
     "-l", "info", "--web-port", "6066", \
     "--without-web"]  # 生产环境可关闭Web UI

apiVersion: apps/v1
kind: Deployment
metadata:
  name: faust-consumer
  namespace: ml-pipeline
spec:
  replicas: 6  # 与Kafka分区数一致
  selector:
    matchLabels:
      app: faust-consumer
  template:
    metadata:
      labels:
        app: faust-consumer
    spec:
      containers:
      - name: consumer
        image: pg-ml-pipeline:1.2.0
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        env:
        - name: POSTGRES_HOST
          value: "postgres.ml-pipeline.svc.cluster.local"
        - name: KAFKA_BROKERS
          value: "kafka-0.kafka-headless:9092,kafka-1.kafka-headless:9092"
        - name: REDIS_HOST
          value: "redis-master.ml-pipeline.svc.cluster.local"
        - name: MLFLOW_TRACKING_URI
          value: "postgresql://ml_user:ml_password@postgres:5432/mlflowdb"
        livenessProbe:
          exec:
            command:
            - python
            - -c
            - "import psycopg2; psycopg2.connect(host='${POSTGRES_HOST}')"
          initialDelaySeconds: 60
          periodSeconds: 30
        readinessProbe:
          tcpSocket:
            port: 6066
          initialDelaySeconds: 30
      nodeSelector:
        workload-type: streaming

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: faust-consumer-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: faust-consumer
  minReplicas: 3
  maxReplicas: 12
  metrics:
  - type: External
    external:
      metric:
        name: kafka_consumergroup_lag
        selector:
          matchLabels:
            group: ml_pipeline_group
            topic: cdc.ecommerce.public.user_behaviors
      target:
        type: AverageValue
        averageValue: "1000"  # 平均每个分区积压1000条消息时扩容

# pgbouncer.ini
[databases]
ecommerce_db = host=postgres port=5432 dbname=ecommerce_db

[pgbouncer]
listen_port = 6432
listen_addr = *
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt

# 连接池模式
pool_mode = transaction  # 事务级，适合短连接

# 连接数配置
default_pool_size = 25
max_client_conn = 1000
reserve_pool_size = 5
reserve_pool_timeout = 3

# 超时设置
server_idle_timeout = 600
server_lifetime = 3600

-- 复合索引优化特征查询
CREATE INDEX CONCURRENTLY idx_features_user_lookup 
ON user_features_realtime (user_id, feature_name) 
INCLUDE (feature_value)  -- PostgreSQL 11+支持INCLUDE

-- 部分索引（只索引热门特征）
CREATE INDEX CONCURRENTLY idx_hot_features
ON user_features_realtime (feature_value)
WHERE feature_name IN ('count_view_last_1h', 'membership_level');

-- BRIN索引（用于时间序列）
CREATE INDEX CONCURRENTLY idx_behavior_time_brin
ON user_behaviors USING BRIN (behavior_timestamp)
WITH (pages_per_range = 128);

# src/monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 定义指标
EVENTS_PROCESSED = Counter(
    'pipeline_events_processed_total',
    '处理的事件总数',
    ['behavior_type', 'status']
)

FEATURE_CALCULATION_TIME = Histogram(
    'pipeline_feature_calculation_seconds',
    '特征计算耗时',
    ['feature_name']
)

KAFKA_LAG = Gauge(
    'pipeline_kafka_lag',
    'Kafka消费延迟',
    ['partition']
)

DB_POOL_SIZE = Gauge(
    'pipeline_db_pool_connections',
    '数据库连接池大小',
    ['pool_type']
)

def monitor_execution(func):
    """装饰器：监控函数执行"""
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            status = 'success'
            return result
        except Exception as e:
            status = 'error'
            raise e
        finally:
            duration = time.time() - start
            FEATURE_CALCULATION_TIME.labels(
                feature_name=func.__name__
            ).observe(duration)
    return wrapper

# 在代码中使用
@monitor_execution
async def calculate_behavior_count(...):
    # 原逻辑
    pass

{
  "dashboard": {
    "title": "ML Pipeline Real-time Monitor",
    "panels": [
      {
        "title": "Events/sec by Type",
        "targets": [{
          "expr": "rate(pipeline_events_processed_total[5m])",
          "legendFormat": "{{behavior_type}}"
        }],
        "type": "graph"
      },
      {
        "title": "Feature Calculation P99",
        "targets": [{
          "expr": "histogram_quantile(0.99, rate(pipeline_feature_calculation_seconds_bucket[5m]))",
          "legendFormat": "{{feature_name}}"
        }]
      },
      {
        "title": "Kafka Consumer Lag",
        "targets": [{
          "expr": "pipeline_kafka_lag",
          "thresholds": {"mode": "absolute", "steps": [{"value": 1000, "color": "red"}]}
        }]
      }
    ]
  }
}

# 在Faust应用中注册关闭钩子
import signal
import sys

def graceful_shutdown(signum, frame):
    """处理SIGTERM信号"""
    logger.info("收到关闭信号，开始优雅停机...")
    
    # I. 停止消费新消息
    app.consumer.unsubscribe()
    
    # II. 等待当前批次处理完成
    asyncio.run(app.stop())
    
    # III. 关闭连接池
    pool.close_all()
    
    logger.info("停机完成")
    sys.exit(0)

signal.signal(signal.SIGTERM, graceful_shutdown)

-- 使用INSERT ON CONFLICT实现幂等
CREATE UNIQUE INDEX CONCURRENTLY idx_unique_feature 
ON user_features_realtime (user_id, feature_name, updated_at);

-- 写入SQL
INSERT INTO user_features_realtime (...) 
VALUES (...)
ON CONFLICT (user_id, feature_name, updated_at) 
DO NOTHING;  -- 已存在则跳过

#!/bin/bash
# scripts/startup.sh

set -e

echo "====== PostgreSQL+Python实时数据管道启动脚本 ======"

# I. 检查环境
check_env() {
    command -v docker >/dev/null 2>&1 || { echo "Docker未安装"; exit 1; }
    command -v docker-compose >/dev/null 2>&1 || { echo "docker-compose未安装"; exit 1; }
    python3.10 --version >/dev/null 2>&1 || { echo "Python 3.10未安装"; exit 1; }
}

# II. 启动基础设施
start_infra() {
    echo "启动PostgreSQL、Kafka、Debezium..."
    docker-compose up -d postgres kafka zookeeper debezium
    
    # 等待服务就绪
    echo "等待PostgreSQL启动..."
    until docker exec pg_ml_pipeline pg_isready; do
        sleep 1
    done
    
    echo "等待Kafka启动..."
    sleep 10
    
    # 创建Kafka Topic
    docker exec kafka kafka-topics --create \
        --topic cdc.ecommerce.public.user_behaviors \
        --bootstrap-server localhost:9092 \
        --partitions 12 \
        --replication-factor 1
        
    echo "基础设施启动完成"
}

# III. 初始化数据库
init_db() {
    echo "初始化数据库Schema..."
    psql postgresql://ml_user:ml_password@localhost:5432/ecommerce_db < scripts/init.sql
    
    # 创建Debezium用户
    psql postgresql://ml_user:ml_password@localhost:5432/ecommerce_db -c "
        CREATE USER debezium WITH PASSWORD 'dbz_p@ssw0rd' REPLICATION;
        GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
    "
}

# IV. 启动Python应用
start_app() {
    echo "启动Faust消费者..."
    source venv/bin/activate
    
    # 启动6个worker
    for i in {1..6}; do
        faust -A src.consumers.behavior_consumer worker \
            -l info \
            --web-port $((6066 + i)) \
            --workdir /tmp/faust_$i &
    done
    
    echo "Faust消费者已后台启动"
}

# 执行流程
check_env
start_infra
init_db
start_app

echo "====== 启动完成！访问 http://localhost:6066 查看Faust监控 ======"