实验数据分析流水线构建：从ETL到报告的全流程优化

第一章：数据流水线架构设计哲学

1.1 现代数据栈的演进与挑战

在数字化转型的浪潮中，实验驱动决策已成为企业增长的核心引擎。从产品经理的UI迭代到算法工程师的模型优化，每一次决策背后都需要可靠的数据支撑。然而，许多团队仍面临数据孤岛、分析延迟、指标不一致等痛点。构建标准化的实验数据分析流水线，正是解决这些问题的关键。

传统工作方式的困境体现在：

现代数据流水线的核心价值：

1.2 罗马数字分层架构体系

构建稳健的流水线需要遵循清晰的分层原则：

Ⅰ. 数据源层（Data Sources）

• 结构化数据：MySQL业务数据库、PostgreSQL交易数据
• 半结构化数据：JSON日志、API响应
• 非结构化数据：用户反馈文本、图像数据

Ⅱ. 采集层（Ingestion）

• 批量采集：Sqoop、DataX
• 流式采集：Kafka、Flink CDC
• 变更数据捕获：Debezium、Maxwell

Ⅲ. 存储层（Storage）

• 对象存储：S3、OSS（原始数据归档）
• 数据湖：Iceberg、Hudi（支持ACID）
• 数据仓库：Snowflake、BigQuery、Redshift

Ⅳ. 转换层（Transformation）

• ETL工具：dbt（首选）、Spark SQL
• 计算引擎：Spark、Presto
• 调度系统：Airflow、Dagster

Ⅴ. 服务层（Serving）

• BI工具：Superset、Tableau
• 特征存储：Feast、Tecton
• API服务：FastAPI、GraphQL

Ⅵ. 治理层（Governance）

• 数据质量：Great Expectations
• 血缘追踪：OpenLineage、Amundsen
• 访问控制：Ranger、IAM

第二章：ETL流程的深度实现与优化

2.1 从业务系统到数据湖的摄取策略

实验数据往往分散在多个异构系统中。构建健壮的摄取层是整个流水线的基石。

增量同步的三种模式对比：

代码实现：基于Debezium的MySQL CDC同步

# docker-compose.yml - Debezium + Kafka集群
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
  
  kafka:
    image: confluentinc/cp-kafka:7.4.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  
  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: debezium
      MYSQL_USER: debezium
      MYSQL_PASSWORD: dbz
    ports:
      - "3306:3306"
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
  
  connect:
    image: debezium/connect:2.5
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
    depends_on:
      - kafka
      - mysql

# register_mysql_connector.py - 注册CDC连接器
import requests
import json

def register_connector():
    """注册MySQL CDC连接器到Debezium"""
    
    config = {
        "name": "experimentdb-connector",
        "config": {
            "connector.class": "io.debezium.connector.mysql.MySqlConnector",
            "tasks.max": "1",
            "database.hostname": "mysql",
            "database.port": "3306",
            "database.user": "debezium",
            "database.password": "dbz",
            "database.server.id": "184054",
            "database.server.name": "experimentdb",
            "database.include.list": "experiments",
            "table.include.list": "experiments.users,experiments.ab_tests",
            "column.include.list": "experiments.users.id,experiments.users.group_id",
            "snapshot.mode": "initial",  # 首次全量，之后增量
            "include.schema.changes": "false",
            "transforms": "route",
            "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
            "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
            "transforms.route.replacement": "$3",
            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
            "value.converter": "io.confluent.connect.avro.AvroConverter",
            "value.converter.schema.registry.url": "http://schema-registry:8081"
        }
    }
    
    response = requests.post(
        "http://localhost:8083/connectors",
        headers={"Content-Type": "application/json"},
        data=json.dumps(config)
    )
    
    if response.status_code == 201:
        print("✅ CDC连接器注册成功")
    else:
        print(f"❌ 注册失败: {response.text}")

if __name__ == "__main__":
    register_connector()

代码深度解析：

1. 快照模式策略：snapshot.mode=initial 确保首次部署时先全量同步当前数据，之后无缝切换到增量binlog读取。这比手动处理全量+增量要可靠得多，避免了数据断层。

2. 主题路由配置：RegexRouter转换器将Debezium默认的server.database.table三层结构简化为仅表名，下游消费者无需关心源数据库结构。

3. 列级过滤：column.include.list显式指定需要同步的列，避免敏感字段（如password）进入数据湖，符合最小权限原则。

4. 独立Schema Registry：Avro格式配合Schema Registry实现schema版本管理，当MySQL表结构变更时，下游系统能自动感知并适配。

2.2 数据质量校验与异常处理

脏数据是分析错误的根源。在ETL阶段必须建立多层防御体系。

# data_quality_checks.py - 数据质量校验框架
from datetime import datetime, timedelta
import pandas as pd
from great_expectations import DataContext
from great_expectations.dataset import PandasDataset

class ExperimentDataValidator:
    """实验数据专用校验器"""
    
    def __init__(self, context_root_dir: str):
        self.context = DataContext(context_root_dir)
    
    def validate_user_assignment(self, df: pd.DataFrame):
        """验证用户分组数据完整性"""
        
        # 转换为Great Expectations数据集
        ge_df = PandasDataset(df)
        
        # Ⅰ. 基础完整性检查
        expect_results = []
        expect_results.append(ge_df.expect_column_values_to_not_be_null("user_id"))
        expect_results.append(ge_df.expect_column_values_to_be_unique("user_id"))
        expect_results.append(ge_df.expect_column_values_to_not_be_null("experiment_id"))
        expect_results.append(ge_df.expect_column_values_to_not_be_null("group_name"))
        
        # Ⅱ. 业务逻辑验证
        # 分组名称只能是control或treatment
        expect_results.append(
            ge_df.expect_column_values_to_be_in_set(
                "group_name",
                ["control", "treatment", "treatment_2"]  # 支持多实验组
            )
        )
        
        # 分配时间必须在实验周期内
        experiment_start = datetime(2024, 1, 1)
        experiment_end = datetime(2024, 12, 31)
        
        expect_results.append(
            ge_df.expect_column_values_to_be_between(
                "assignment_time",
                min_value=experiment_start,
                max_value=experiment_end,
                parse_strings_as_datetimes=True
            )
        )
        
        # Ⅲ. 统计分布验证
        # 检查分组比例是否失衡（如90%进入实验组）
        group_counts = df['group_name'].value_counts()
        control_ratio = group_counts.get('control', 0) / len(df)
        
        if not (0.4 <= control_ratio <= 0.6):
            expect_results.append({
                "success": False,
                "expectation_type": "group_balance_check",
                "message": f"对照组比例异常: {control_ratio:.2%}"
            })
        
        # Ⅳ. 生成质量报告
        success_rate = sum(r.get("success", False) for r in expect_results) / len(expect_results)
        
        return {
            "validation_time": datetime.utcnow().isoformat(),
            "total_checks": len(expect_results),
            "passed_checks": sum(r.get("success", False) for r in expect_results),
            "success_rate": success_rate,
            "detailed_results": expect_results,
            "failed_expectations": [
                r for r in expect_results 
                if not r.get("success", False)
            ]
        }

# 使用示例
validator = ExperimentDataValidator("/path/to/great_expectations")

# 模拟从Kafka消费的数据
sample_assignment_data = pd.DataFrame({
    'user_id': ['u1', 'u2', 'u3', 'u4'],
    'experiment_id': ['exp_001'] * 4,
    'group_name': ['control', 'treatment', 'control', 'treatment'],
    'assignment_time': pd.date_range('2024-01-15', periods=4)
})

quality_report = validator.validate_user_assignment(sample_assignment_data)
print(json.dumps(quality_report, indent=2))

2.3 Mermaid流程总结

第三章：数据仓库建模与dbt工程化实践

3.1 实验数据仓库的维度建模

实验分析最适合采用星型模型，以实验事实表为核心，关联用户、产品、时间等维度。

核心表结构设计：

-- models/staging/stg_user_assignments.sql
/* 用户分组事实表 - dbt基础模型 */
{{ config(
    materialized='incremental',
    unique_key='user_assignment_id',
    incremental_strategy='merge',
    partition_by={
      "field": "assignment_date",
      "data_type": "date"
    }
) }}

WITH source_data AS (
    SELECT 
        -- 主键生成策略：确保幂等性
        MD5(CONCAT(user_id, experiment_id, assignment_time)) AS user_assignment_id,
        user_id,
        experiment_id,
        experiment_name,
        group_name,
        
        -- 时间维度代理键
        CAST(assignment_time AS DATE) AS assignment_date,
        
        -- 元数据审计字段
        '{{ invocation_id }}' AS dbt_run_id,
        '{{ run_started_at }}' AS processed_at
        
    FROM {{ source('raw', 'user_assignments') }}
    
    {% if is_incremental() %}
    -- 增量加载：只处理新数据
    WHERE assignment_time > (SELECT MAX(assignment_time) FROM {{ this }})
    {% endif %}
)

SELECT * FROM source_data

深度解析dbt配置：

1. 增量策略选择：incremental_strategy='merge' 在BigQuery/Snowflake上效率最高，相比delete+insert模式，merge减少了扫描量，特别适合每日百万级用户分流的场景。

2. 分区字段设计：按assignment_date分区后，查询特定实验周期内的数据只需扫描对应分区，成本降低90%以上。例如查询7天实验数据，BigQuery仅扫描7个分区而非全表。

3. 幂等主键：使用MD5哈希生成主键，确保即使重复运行也不会产生重复记录。这在Airflow任务重试时至关重要，避免了数据重复计算。

4. 元数据追踪：dbt_run_id和processed_at字段为每条数据打上处理标记，便于问题排查和数据血缘追踪。

3.2 实验指标体系的标准化

定义企业级的实验指标是数据治理的关键。我们通过dbt的宏（macro）实现指标的统一计算。

-- macros/calculate_conversion_rate.sql
/* 标准化转化率计算宏 */
{% macro calculate_conversion_rate(
    event_table, 
    base_table, 
    user_id_column='user_id',
    conversion_event='purchase',
    date_spine='assignment_date',
    experiment_grain=['experiment_id', 'group_name']
) %}

WITH base_population AS (
    -- 基准用户群：实验组分配用户
    SELECT
        {% for grain in experiment_grain %}
        {{ grain }},{% endfor %}
        {{ date_spine }},
        COUNT(DISTINCT {{ user_id_column }}) AS total_users
    FROM {{ base_table }}
    GROUP BY {% for grain in experiment_grain %}{{ grain }}{% if not loop.last %},{% endif %}{% endfor %}, {{ date_spine }}
),

conversion_events AS (
    -- 转化事件：在实验窗口内完成指定行为
    SELECT
        e.{{ user_id_column }},
        {% for grain in experiment_grain %}
        b.{{ grain }},{% endfor %}
        b.{{ date_spine }},
        COUNT(*) AS conversion_count
    FROM {{ event_table }} e
    INNER JOIN {{ base_table }} b 
        ON e.{{ user_id_column }} = b.{{ user_id_column }}
    WHERE e.event_name = '{{ conversion_event }}'
        AND e.event_time BETWEEN 
            b.assignment_time AND 
            DATE_ADD(b.assignment_time, INTERVAL b.experiment_duration DAY)
    GROUP BY e.{{ user_id_column }}, {% for grain in experiment_grain %}b.{{ grain }}{% if not loop.last %},{% endif %}{% endfor %}, b.{{ date_spine }}
)

SELECT
    bp.{% for grain in experiment_grain %}{{ grain }}{% if not loop.last %},{% endif %}{% endfor %},
    bp.{{ date_spine }},
    bp.total_users,
    COUNT(DISTINCT ce.{{ user_id_column }}) AS converted_users,
    ROUND(
        COUNT(DISTINCT ce.{{ user_id_column }}) * 100.0 / bp.total_users, 
        4
    ) AS conversion_rate_pct
FROM base_population bp
LEFT JOIN conversion_events ce 
    ON {% for grain in experiment_grain %}bp.{{ grain }} = ce.{{ grain }}{% if not loop.last %} AND {% endif %}{% endfor %}
GROUP BY bp.{% for grain in experiment_grain %}{{ grain }}{% if not loop.last %},{% endif %}{% endfor %}, bp.{{ date_spine }}, bp.total_users

{% endmacro %}

调用示例：

-- models/marts/experiment_conversion_rates.sql
{{ config(materialized='table') }}

WITH experiment_base AS (
    SELECT 
        user_id,
        experiment_id,
        group_name,
        assignment_date,
        assignment_time,
        experiment_duration
    FROM {{ ref('fct_user_assignments') }}
    WHERE experiment_status = 'active'
),

purchase_events AS (
    SELECT * FROM {{ ref('stg_events') }}
    WHERE event_name = 'purchase'
)

{{ calculate_conversion_rate(
    event_table='purchase_events',
    base_table='experiment_base',
    conversion_event='purchase',
    experiment_grain=['experiment_id', 'group_name']
) }}

3.3 数据血缘与文档自动化工

-- dbt_project.yml
name: 'experiment_analytics'
version: '1.0.0'
config-version: 2

# 模型路径配置
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]

# 关键配置：自动生成文档和血缘
docs-paths: ["docs"]

models:
  experiment_analytics:
    +materialized: view
    +persist_docs:
      relation: true
      columns: true
    
    staging:
      +schema: stg
      +tags: ["daily", "raw"]
    
    marts:
      +schema: analytics
      +tags: ["business_critical"]
      +post-hook:
        - "GRANT SELECT ON {{ this }} TO ROLE analyst"

第四章：Airflow工作流编排与生产化部署

4.1 动态DAG生成模式

手动编写每个实验的DAG不可扩展。我们采用元数据驱动的方式自动生成工作流。

# dags/dag_factory.py
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta
import yaml
import os

def create_experiment_dag(experiment_config):
    """根据实验配置动态生成DAG"""
    
    dag_id = f"experiment_pipeline_{experiment_config['id']}"
    
    default_args = {
        'owner': 'data_team',
        'depends_on_past': False,
        'email_on_failure': True,
        'email_on_retry': False,
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
        'execution_timeout': timedelta(hours=2)
    }
    
    with DAG(
        dag_id=dag_id,
        default_args=default_args,
        description=f"实验 {experiment_config['name']} 数据处理流水线",
        schedule_interval=experiment_config['schedule'],
        start_date=datetime(2024, 1, 1),
        catchup=False,
        tags=['experiment', experiment_config['team']],
        max_active_runs=1,
        # 启用SLA监控
        sla_miss_callback=notify_slack_channel,
    ) as dag:
        
        start = DummyOperator(task_id='start')
        
        # Ⅰ. 数据摄取阶段
        with TaskGroup('ingestion') as ingestion_tg:
            sync_user_data = PostgresOperator(
                task_id='sync_user_assignments',
                postgres_conn_id='experiment_db',
                sql="""
                    COPY stg_user_assignments 
                    FROM 's3://experiment-data/assignments/{{ ds }}'
                    IAM_ROLE 'arn:aws:iam::1234567890:role/RedshiftCopyRole'
                    FORMAT AS PARQUET;
                """
            )
            
            sync_events = PostgresOperator(
                task_id='sync_events',
                postgres_conn_id='experiment_db',
                sql="""
                    COPY stg_events 
                    FROM 's3://experiment-data/events/{{ ds }}'
                    IAM_ROLE 'arn:aws:iam::1234567890:role/RedshiftCopyRole'
                    FORMAT AS JSON 'auto';
                """
            )
        
        # Ⅱ. dbt转换阶段
        with TaskGroup('transformation') as transform_tg:
            dbt_run = BashOperator(
                task_id='dbt_run_models',
                bash_command=f"""
                    cd /opt/dbt/experiment_analytics &&
                    dbt run --select tag:{experiment_config['id']} 
                    --target prod --profiles-dir ./
                """
            )
            
            dbt_test = BashOperator(
                task_id='dbt_data_tests',
                bash_command=f"""
                    cd /opt/dbt/experiment_analytics &&
                    dbt test --select tag:{experiment_config['id']}
                    --store-failures
                """
            )
            
            dbt_run >> dbt_test
        
        # Ⅲ. 分析计算阶段
        calculate_metrics = PythonOperator(
            task_id='calculate_experiment_metrics',
            python_callable=compute_statistical_significance,
            op_kwargs={
                'experiment_id': experiment_config['id'],
                'metrics': experiment_config['primary_metrics'],
                'alpha': experiment_config.get('significance_level', 0.05)
            }
        )
        
        # Ⅳ. 报告生成阶段
        generate_report = PythonOperator(
            task_id='generate_experiment_report',
            python_callable=create_html_report,
            op_kwargs={
                'experiment_id': experiment_config['id'],
                'template': experiment_config['report_template']
            }
        )
        
        # Ⅴ. 质量监控阶段
        data_quality_check = GreatExpectationsOperator(
            task_id='data_quality_check',
            data_context_root_dir="/opt/great_expectations",
            expectation_suite_name=f"{experiment_config['id']}_suite",
            batch_kwargs={
                'table': 'fct_user_assignments',
                'schema': 'analytics'
            }
        )
        
        end = DummyOperator(task_id='end')
        
        # 依赖关系
        start >> ingestion_tg >> transform_tg >> data_quality_check 
        data_quality_check >> calculate_metrics >> generate_report >> end
        
        return dag

# 扫描配置文件目录自动生成DAG
config_dir = "/opt/airflow/dags/configs"
for filename in os.listdir(config_dir):
    if filename.endswith('.yaml'):
        with open(os.path.join(config_dir, filename)) as f:
            config = yaml.safe_load(f)
            globals()[f"dag_{config['id']}"] = create_experiment_dag(config)

配置示例：

# configs/recommendation_exp.yaml
id: "rec_algo_v2"
name: "推荐算法2.0 A/B测试"
team: "ml_platform"
schedule: "0 6 * * *"  # 每天6点运行
primary_metrics:
  - ctr
  - dwell_time
  - conversion_rate
significance_level: 0.05
report_template: "ml_experiment.html"

4.2 生产环境部署配置

# Dockerfile - Airflow生产镜像
FROM apache/airflow:2.7.3-python3.11

USER root
RUN apt-get update && apt-get install -y \
    build-essential \
    libpq-dev \
    git \
    && rm -rf /var/lib/apt/lists/*

USER airflow

# 安装依赖
COPY requirements.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements.txt

# 复制DAG和配置
COPY --chown=airflow:airflow dags/ /opt/airflow/dags/
COPY --chown=airflow:airflow configs/ /opt/airflow/dags/configs/
COPY --chown=airflow:airflow plugins/ /opt/airflow/plugins/

# 环境变量配置
ENV AIRFLOW__CORE__EXECUTOR=CeleryExecutor
ENV AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:password@postgres:5432/airflow
ENV AIRFLOW__CELERY__BROKER_URL=redis://redis:6379/0
ENV AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:password@postgres:5432/airflow
ENV AIRFLOW__WEBSERVER__WORKERS_CLASS=gevent

# docker-compose.prod.yml
version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow_password
      POSTGRES_DB: airflow
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    
  airflow-webserver:
    build: .
    command: webserver
    ports:
      - "8080:8080"
    environment:
      - AIRFLOW__WEBSERVER__SECRET_KEY=prod_secret_key_here
    depends_on:
      - postgres
      - redis
  
  airflow-scheduler:
    build: .
    command: scheduler
    depends_on:
      - postgres
      - redis
  
  airflow-worker:
    build: .
    command: celery worker
    environment:
      - AIRFLOW__CELERY__WORKER_CONCURRENCY=16
    depends_on:
      - postgres
      - redis

volumes:
  postgres_data:

4.3 Mermaid调度架构

第五章：统计分析层与显著性检验

5.1 贝叶斯框架下的实验评估

传统频率学派p值方法存在诸多局限。现代实验平台普遍采用贝叶斯方法，提供更直观的决策依据。

# analysis/bayesian_analysis.py
import numpy as np
from scipy import stats
import pymc as pm
import arviz as az
import pandas as pd

class BayesianExperimentAnalyzer:
    """贝叶斯实验分析引擎"""
    
    def __init__(self, alpha_prior=2.0, beta_prior=2.0):
        """初始化Beta先验分布"""
        self.alpha_prior = alpha_prior
        self.beta_prior = beta_prior
    
    def analyze_conversion_rate(
        self, 
        control_conversions: int, 
        control_users: int,
        treatment_conversions: int,
        treatment_users: int,
        mcmc_draws: int = 10000
    ):
        """分析转化率的贝叶斯模型"""
        
        # Ⅰ. 构建概率模型
        with pm.Model() as conversion_model:
            # 先验分布：Beta分布
            theta_control = pm.Beta(
                'theta_control', 
                alpha=self.alpha_prior, 
                beta=self.beta_prior
            )
            theta_treatment = pm.Beta(
                'theta_treatment', 
                alpha=self.alpha_prior, 
                beta=self.beta_prior
            )
            
            # 似然函数：二项分布
            obs_control = pm.Binomial(
                'obs_control',
                n=control_users,
                p=theta_control,
                observed=control_conversions
            )
            obs_treatment = pm.Binomial(
                'obs_treatment',
                n=treatment_users,
                p=theta_treatment,
                observed=treatment_conversions
            )
            
            # 相对提升率
            lift = pm.Deterministic(
                'lift', 
                (theta_treatment - theta_control) / theta_control
            )
            
            # 执行MCMC采样
            trace = pm.sample(
                mcmc_draws,
                chains=4,
                cores=2,
                return_inferencedata=True,
                random_seed=42
            )
        
        # Ⅱ. 计算后验统计量
        posterior_summary = az.summary(trace, hdi_prob=0.95)
        
        # Ⅲ. 决策指标
        lift_samples = trace.posterior['lift'].values.flatten()
        
        results = {
            "model_type": "bayesian_beta_binomial",
            "posterior": {
                "control_rate": {
                    "mean": float(trace.posterior['theta_control'].mean()),
                    "hdi_95_lower": float(az.hdi(trace, var_names=['theta_control'])['theta_control'][0]),
                    "hdi_95_upper": float(az.hdi(trace, var_names=['theta_control'])['theta_control'][1])
                },
                "treatment_rate": {
                    "mean": float(trace.posterior['theta_treatment'].mean()),
                    "hdi_95_lower": float(az.hdi(trace, var_names=['theta_treatment'])['theta_treatment'][0]),
                    "hdi_95_upper": float(az.hdi(trace, var_names=['theta_treatment'])['theta_treatment'][1])
                }
            },
            "decision_metrics": {
                "probability_of_beat_control": float(np.mean(lift_samples > 0)),
                "expected_lift": float(np.mean(lift_samples)),
                "lift_hdi_95_lower": float(np.percentile(lift_samples, 2.5)),
                "lift_hdi_95_upper": float(np.percentile(lift_samples, 97.5)),
                "risk_if_launched": float(np.mean(np.where(lift_samples < 0, abs(lift_samples), 0)))
            }
        }
        
        return results

# 批量分析函数
def analyze_all_experiments(engine, execution_date):
    """分析当日所有活跃实验"""
    
    query = """
        SELECT 
            experiment_id,
            group_name,
            SUM(converted_users) as conversions,
            SUM(total_users) as users
        FROM analytics.experiment_conversion_rates
        WHERE assignment_date = :exec_date
        GROUP BY experiment_id, group_name
    """
    
    df = pd.read_sql(query, engine, params={'exec_date': execution_date})
    
    # 按实验聚合
    experiments = {}
    for exp_id in df['experiment_id'].unique():
        exp_data = df[df['experiment_id'] == exp_id]
        
        control = exp_data[exp_data['group_name'] == 'control'].iloc[0]
        treatment = exp_data[exp_data['group_name'] == 'treatment'].iloc[0]
        
        analyzer = BayesianExperimentAnalyzer()
        result = analyzer.analyze_conversion_rate(
            control_conversions=control['conversions'],
            control_users=control['users'],
            treatment_conversions=treatment['conversions'],
            treatment_users=treatment['users']
        )
        
        experiments[exp_id] = result
    
    return experiments

5.2 连续型指标的贝叶斯建模

对于收入、时长等连续指标，采用伽马-正态分层模型：

# analysis/continuous_metrics.py
class ContinuousMetricAnalyzer:
    """连续型指标分析（如收入、时长）"""
    
    def analyze_revenue(
        self,
        control_revenues: np.ndarray,
        treatment_revenues: np.ndarray
    ):
        """基于T检验的贝叶斯版本"""
        
        with pm.Model() as revenue_model:
            # 分层先验
            mu_prior = pm.Normal('mu_prior', mu=0, sigma=100)
            sigma_prior = pm.HalfNormal('sigma_prior', sigma=50)
            
            # 组间差异
            delta = pm.Normal('delta', mu=0, sigma=10)
            
            # 组内均值
            mu_control = pm.Normal('mu_control', mu=mu_prior, sigma=sigma_prior)
            mu_treatment = pm.Deterministic('mu_treatment', mu_control + delta)
            
            # 精度（方差的倒数）
            tau_control = pm.Gamma('tau_control', alpha=2, beta=1)
            tau_treatment = pm.Gamma('tau_treatment', alpha=2, beta=1)
            
            # 似然
            obs_control = pm.Normal(
                'obs_control',
                mu=mu_control,
                tau=tau_control,
                observed=control_revenues
            )
            
            obs_treatment = pm.Normal(
                'obs_treatment',
                mu=mu_treatment,
                tau=tau_treatment,
                observed=treatment_revenues
            )
            
            # 采样
            trace = pm.sample(5000, chains=4, cores=2)
        
        return trace

# 性能优化：使用向量化操作
def vectorized_credible_interval(posterior_samples, alpha=0.05):
    """快速计算可信区间"""
    lower = np.percentile(posterior_samples, 100 * alpha / 2)
    upper = np.percentile(posterior_samples, 100 * (1 - alpha / 2))
    return (lower, upper)

5.3 Mermaid分析流程

第六章：可视化与报告自动化系统

6.1 Superset二次开发：实验看板

原生BI工具无法满足实验分析的特殊需求（如P值展示、分组对比）。我们通过Superset的Jinja模板和自定义可视化插件扩展功能。

# superset_custom_vis/plugins/experiment_chart.py
from superset.viz import BaseViz
from flask import request
import json

class ExperimentBayesianChart(BaseViz):
    """自定义实验分析图表"""
    
    viz_type = "experiment_bayesian"
    verbose_name = "Experiment Bayesian Analysis"
    is_timeseries = False
    
    def query_obj(self):
        """构建查询对象"""
        d = super().query_obj()
        
        # 强制添加实验ID过滤器
        form_data = self.form_data
        experiment_id = form_data.get('experiment_id')
        
        if experiment_id:
            d['filters'].append({
                'col': 'experiment_id',
                'op': '==',
                'val': experiment_id
            })
        
        return d
    
    def get_data(self, df):
        """后处理数据，计算贝叶斯指标"""
        
        if df.empty:
            return {}
        
        # 从数据库获取原始数据
        engine = self.datasource.database.get_sqla_engine()
        
        query = """
            SELECT 
                group_name,
                SUM(conversions) as conv,
                SUM(users) as users
            FROM analytics.experiment_stats
            WHERE experiment_id = :exp_id
            GROUP BY group_name
        """
        
        stats_df = pd.read_sql(
            query, 
            engine, 
            params={'exp_id': self.form_data['experiment_id']}
        )
        
        # 调用贝叶斯分析
        analyzer = BayesianExperimentAnalyzer()
        result = analyzer.analyze_conversion_rate(
            control_conversions=stats_df.loc[stats_df['group_name']=='control', 'conv'].iloc[0],
            control_users=stats_df.loc[stats_df['group_name']=='control', 'users'].iloc[0],
            treatment_conversions=stats_df.loc[stats_df['group_name']=='treatment', 'conv'].iloc[0],
            treatment_users=stats_df.loc[stats_df['group_name']=='treatment', 'users'].iloc[0]
        )
        
        # 转换为ECharts配置
        return {
            "probability_beat_control": result['decision_metrics']['probability_of_beat_control'],
            "expected_lift": result['decision_metrics']['expected_lift'],
            "posterior_plot": self._generate_postior_plot(result)
        }
    
    def _generate_postior_plot(self, bayesian_result):
        """生成后验分布对比图"""
        # 返回ECharts配置
        return {
            "title": {"text": "转化率后验分布"},
            "tooltip": {"trigger": "axis"},
            "legend": {"data": ["Control", "Treatment"]},
            "xAxis": {"type": "value"},
            "yAxis": {"type": "category"},
            "series": [
                {
                    "name": "Control",
                    "type": "bar",
                    "data": bayesian_result['posterior']['control_rate']
                },
                {
                    "name": "Treatment", 
                    "type": "bar",
                    "data": bayesian_result['posterior']['treatment_rate']
                }
            ]
        }

6.2 自动化报告生成引擎

# reporting/report_generator.py
from jinja2 import Environment, FileSystemLoader
import markdown
import weasyprint
import boto3
from datetime import datetime

class ExperimentReportGenerator:
    """实验报告生成器"""
    
    def __init__(self, template_dir="/opt/reporting/templates"):
        self.env = Environment(loader=FileSystemLoader(template_dir))
        self.s3_client = boto3.client('s3')
    
    def generate_html_report(self, experiment_id, results):
        """生成完整的HTML报告"""
        
        template = self.env.get_template('experiment_report.html')
        
        # Ⅰ. 报告元数据
        report_meta = {
            "experiment_id": experiment_id,
            "generated_at": datetime.utcnow().isoformat(),
            "analysis_model": "贝叶斯Beta-二项模型",
            "confidence_level": "95% HDI",
            "data_period": f"{results['start_date']} to {results['end_date']}"
        }
        
        # Ⅱ. 关键发现表格
        findings = self._format_key_findings(results)
        
        # Ⅲ. 技术细节表格
        technical_details = self._format_technical_details(results)
        
        # Ⅳ. 渲染HTML
        html_content = template.render(
            meta=report_meta,
            findings=findings,
            technical_details=technical_details,
            posterior_plot=results['plots']['posterior'],
            trace_plot=results['plots']['trace']
        )
        
        return html_content
    
    def _format_key_findings(self, results):
        """格式化关键发现"""
        
        decision_matrix = {
            "probability_of_beat_control": {
                "value": f"{results['decision_metrics']['probability_of_beat_control']:.2%}",
                "interpretation": "实验组优于对照组的概率",
                "threshold": "> 95% 推荐上线",
                "status": "✅ PASS" if results['decision_metrics']['probability_of_beat_control'] > 0.95 else "⚠️ REVIEW"
            },
            "expected_lift": {
                "value": f"{results['decision_metrics']['expected_lift']:.2%}",
                "interpretation": "预期相对提升",
                "threshold": "> 2% 有意义",
                "status": "✅ PASS" if results['decision_metrics']['expected_lift'] > 0.02 else "❌ FAIL"
            },
            "risk_if_launched": {
                "value": f"{results['decision_metrics']['risk_if_launched']:.2%}",
                "interpretation": "上线后的下行风险",
                "threshold": "< 1% 可接受",
                "status": "✅ PASS" if results['decision_metrics']['risk_if_launched'] < 0.01 else "❌ FAIL"
            }
        }
        
        return decision_matrix
    
    def _format_technical_details(self, results):
        """格式化技术细节"""
        
        return {
            "样本量统计": {
                "对照组用户数": f"{results['sample_sizes']['control']:,}",
                "实验组用户数": f"{results['sample_sizes']['treatment']:,}",
                "数据比例": f"{results['sample_sizes']['treatment'] / results['sample_sizes']['control']:.2f}"
            },
            "模型诊断": {
                "R-hat统计量": f"{results['model_diagnostics']['r_hat']:.3f}",
                "有效样本量": f"{results['model_diagnostics']['ess']:,}",
                "MCMC采样收敛": "✅ 收敛" if results['model_diagnostics']['r_hat'] < 1.1 else "❌ 未收敛"
            }
        }
    
    def publish_report(self, experiment_id, html_content, bucket="experiment-reports"):
        """发布报告到S3并返回URL"""
        
        s3_key = f"{experiment_id}/{datetime.utcnow().strftime('%Y-%m-%d')}/report.html"
        
        self.s3_client.put_object(
            Bucket=bucket,
            Key=s3_key,
            Body=html_content,
            ContentType='text/html',
            ACL='public-read'
        )
        
        url = f"https://{bucket}.s3.amazonaws.com/{s3_key}"
        return url

第七章：性能优化与成本治理

7.1 查询性能优化策略

Ⅰ. 分区裁剪（Partition Pruning）

-- 优化前：全表扫描500TB
SELECT * FROM events WHERE event_time > '2024-01-01';

-- 优化后：仅扫描7个分区（约3.5TB）
SELECT * FROM events 
WHERE event_date BETWEEN '2024-01-01' AND '2024-01-07'
  AND event_time > '2024-01-01';

Ⅱ. 物化视图策略

-- 为高频查询创建物化视图
CREATE MATERIALIZED VIEW mv_daily_experiment_stats AS
SELECT 
    experiment_id,
    group_name,
    assignment_date,
    COUNT(DISTINCT user_id) as users,
    SUM(conversions) as conversions
FROM analytics.fct_user_assignments
GROUP BY experiment_id, group_name, assignment_date;

-- 自动刷新设置
CREATE OR REPLACE TASK refresh_mv_stats
    WAREHOUSE = COMPUTE_WH
    SCHEDULE = 'USING CRON 0 6 * * * UTC'  -- 每天6点
AS
    ALTER MATERIALIZED VIEW mv_daily_experiment_stats REFRESH;

Ⅲ. 结果缓存机制

# utils/cache_manager.py
import redis
import pickle
import hashlib
from functools import wraps

def cache_analysis_results(ttl=3600):
    """分析结果缓存装饰器"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            
            # 生成缓存键：基于函数名和参数
            cache_key_base = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            cache_key = hashlib.md5(cache_key_base.encode()).hexdigest()
            
            # 尝试从Redis读取
            redis_client = redis.Redis(host='redis', port=6379)
            cached_result = redis_client.get(cache_key)
            
            if cached_result:
                print(f"✅ Cache hit for {cache_key}")
                return pickle.loads(cached_result)
            
            # 执行原始函数
            result = func(*args, **kwargs)
            
            # 写入缓存
            redis_client.setex(
                cache_key,
                ttl,
                pickle.dumps(result)
            )
            
            return result
        
        return wrapper
    
    return decorator

# 使用示例
@cache_analysis_results(ttl=7200)  # 缓存2小时
def calculate_experiment_metrics(experiment_id, metrics):
    # 耗时计算逻辑
    pass

7.2 成本监控与优化

数据流水线成本分析（月度）：

生命周期策略配置：

// s3_lifecycle_policy.json
{
    "Rules": [
        {
            "ID": "Transition to IA",
            "Status": "Enabled",
            "Filter": {
                "Prefix": "raw_data/"
            },
            "Transitions": [
                {
                    "Days": 90,
                    "StorageClass": "STANDARD_IA"
                }
            ]
        },
        {
            "ID": "Archive Old Data",
            "Status": "Enabled",
            "Filter": {
                "Prefix": "raw_data/"
            },
            "Transitions": [
                {
                    "Days": 365,
                    "StorageClass": "GLACIER"
                }
            ]
        },
        {
            "ID": "Delete Temp Data",
            "Status": "Enabled",
            "Filter": {
                "Prefix": "temp/"
            },
            "Expiration": {
                "Days": 7
            }
        }
    ]
}