数据平台工程：Data Mesh、FinOps 与 DataOps 生产实践

# iceberg_catalog_setup.py
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, StringType, TimestampType, NestedField
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform

def init_domain_catalog(domain_name: str, warehouse_path: str):
    """初始化领域级 Iceberg Catalog，支持多租户隔离"""
    catalog = load_catalog(
        "rest",
        **{
            "uri": "http://iceberg-rest:8181",
            "warehouse": warehouse_path,
            "s3.endpoint": "http://minio:9000",
        }
    )
    
    schema = Schema(
        NestedField(1, "event_id", LongType(), required=True),
        NestedField(2, "domain", StringType(), required=True),
        NestedField(3, "payload", StringType(), required=False),
        NestedField(4, "event_time", TimestampType(), required=True),
        NestedField(5, "dt", StringType(), required=True),
    )
    
    partition_spec = PartitionSpec(
        PartitionField(source_id=4, field_id=1000, transform=DayTransform(), name="dt")
    )
    
    table = catalog.create_table(
        identifier=f"{domain_name}.events",
        schema=schema,
        partition_spec=partition_spec,
        properties={"write_compression": "ZSTD"}
    )
    return table

if __name__ == "__main__":
    table = init_domain_catalog("order_domain", "s3://data-lake/warehouse")
    print(f"Table created: {table.name()}")

# domain_registration.yaml
apiVersion: dataplatform.io/v1
kind: DomainRegistry
metadata:
  name: order-domain
  labels:
    owner: order-squad
    costCenter: cc-8821
spec:
  description: "订单领域数据产品，涵盖下单、支付、履约全生命周期"
  dataProducts:
    - name: order_events
      layer: raw
      format: iceberg
      updateFrequency: realtime
      sla: p99_latency_5s
      schemaRef: s3://data-lake/schemas/order_events.avsc
      accessPolicy: internal
      consumers:
        - finance_domain
        - logistics_domain

// order_domain_event.proto
syntax = "proto3";
package order.v1;

option java_multiple_files = true;

message OrderCreatedEvent {
  string order_id = 1;
  string user_id = 2;
  double amount = 3;
  string currency = 4;
  int64 created_at_ms = 5;
  repeated OrderItem items = 6;
  
  message OrderItem {
    string sku = 1;
    int32 quantity = 2;
    double unit_price = 3;
  }
}

message OrderPaidEvent {
  string order_id = 1;
  string payment_id = 2;
  PaymentStatus status = 3;
  int64 paid_at_ms = 4;
  
  enum PaymentStatus {
    UNKNOWN = 0;
    SUCCESS = 1;
    FAILED = 2;
    REFUNDED = 3;
  }
}

# data_product.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import hashlib
import json

@dataclass
class DataProduct:
    name: str
    domain: str
    owner: str
    schema_version: str
    port: Dict[str, str]  # input/output ports
    quality_contract: Dict
    lineage_upstream: List[str] = field(default_factory=list)
    
    def get_address(self) -> str:
        """基于领域与产品名生成全局唯一寻址标识"""
        return f"dataproduct://{self.domain}/{self.name}@{self.schema_version}"
    
    def compute_checksum(self) -> str:
        """计算数据产品定义的标准化校验和，用于版本比对"""
        canonical = json.dumps({
            "name": self.name,
            "schema": self.schema_version,
            "ports": sorted(self.port.items()),
            "quality": self.quality_contract
        }, sort_keys=True)
        return hashlib.sha256(canonical.encode()).hexdigest()[:16]
    
    def to_registry_manifest(self) -> dict:
        return {
            "address": self.get_address(),
            "owner": self.owner,
            "registered_at": datetime.utcnow().isoformat(),
            "checksum": self.compute_checksum(),
            "quality_sla": self.quality_contract
        }

# 示例：订单事件数据产品
order_events_dp = DataProduct(
    name="order_events",
    domain="order",
    owner="order-squad@company.com",
    schema_version="2.3.1",
    port={
        "input": "kafka://events.order.raw",
        "output": "iceberg://lake/order/events"
    },
    quality_contract={
        "freshness": "5m",
        "null_rate": "<0.01%",
        "schema_evolution": "backward_compatible"
    },
    lineage_upstream=["mysql://db-primary/orders"]
)

print(order_events_dp.to_registry_manifest())

# federated_policy_engine.py
import re
from typing import Any

class DataPolicy:
    def __init__(self, policy_id: str, domain_scope: str, rule: dict):
        self.policy_id = policy_id
        self.domain_scope = domain_scope  # "*" 表示全局
        self.rule = rule
    
    def evaluate(self, data_product_manifest: dict) -> dict:
        results = {"policy_id": self.policy_id, "passed": True, "violations": []}
        
        if "required_tags" in self.rule:
            missing = set(self.rule["required_tags"]) - set(data_product_manifest.get("tags", []))
            if missing:
                results["passed"] = False
                results["violations"].append(f"Missing tags: {missing}")
        
        if "naming_convention" in self.rule:
            pattern = self.rule["naming_convention"]
            name = data_product_manifest.get("name", "")
            if not re.match(pattern, name):
                results["passed"] = False
                results["violations"].append(f"Name '{name}' does not match pattern '{pattern}'")
        
        if "cost_limit_usd" in self.rule:
            estimated = data_product_manifest.get("estimated_monthly_cost_usd", 0)
            if estimated > self.rule["cost_limit_usd"]:
                results["passed"] = False
                results["violations"].append(f"Cost {estimated} exceeds limit {self.rule['cost_limit_usd']}")
        
        return results

# 全局策略：所有数据产品必须标记 costCenter
# 领域策略：order 域的产品名必须以 order_ 前缀开头
global_policy = DataPolicy(
    policy_id="P001",
    domain_scope="*",
    rule={"required_tags": ["costCenter", "owner", "dataClassification"]}
)

domain_policy = DataPolicy(
    policy_id="P002",
    domain_scope="order",
    rule={"naming_convention": r"^order_.+"}
)

manifest = {
    "name": "order_events",
    "domain": "order",
    "tags": ["costCenter:cc-8821", "owner:order-squad", "dataClassification:internal"]
}

print(global_policy.evaluate(manifest))
print(domain_policy.evaluate(manifest))

# contract_test.py
import json
from jsonschema import validate, ValidationError

class CrossDomainContract:
    def __init__(self, consumer_domain: str, provider_dp_address: str, expected_schema: dict):
        self.consumer_domain = consumer_domain
        self.provider_dp_address = provider_dp_address
        self.expected_schema = expected_schema
    
    def validate_compatibility(self, actual_schema: dict) -> dict:
        """验证实际 schema 是否满足消费者契约"""
        result = {"consumer": self.consumer_domain, "compatible": True, "errors": []}
        try:
            # 这里使用 jsonschema 进行简化演示，生产中应对 Avro/Protobuf 做深度兼容检查
            validate(instance={"field_example": "test"}, schema=self.expected_schema)
            # 检查字段存在性
            required = self.expected_schema.get("required", [])
            actual_props = actual_schema.get("properties", {})
            for field in required:
                if field not in actual_props:
                    result["compatible"] = False
                    result["errors"].append(f"Required field '{field}' missing in provider schema")
        except ValidationError as e:
            result["compatible"] = False
            result["errors"].append(str(e))
        return result

# finance 域消费 order_events，要求必须包含 order_id 与 amount
finance_contract = CrossDomainContract(
    consumer_domain="finance",
    provider_dp_address="dataproduct://order/order_events@2.3.1",
    expected_schema={
        "type": "object",
        "required": ["order_id", "amount", "currency"],
        "properties": {
            "order_id": {"type": "string"},
            "amount": {"type": "number"},
            "currency": {"type": "string"}
        }
    }
)

provider_actual = {
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "amount": {"type": "number"}
        # currency 缺失，将导致兼容检查失败
    }
}

print(finance_contract.validate_compatibility(provider_actual))

# data_product_layers.py
from enum import Enum
from dataclasses import dataclass

class DataLayer(Enum):
    RAW = "raw"
    CLEANED = "cleaned"
    AGGREGATED = "aggregated"
    SERVING = "serving"

@dataclass
class LayerSLO:
    freshness_minutes: int
    completeness_pct: float
    uniqueness_check: bool
    schema_drift_alert: bool

LAYER_DEFAULTS = {
    DataLayer.RAW: LayerSLO(freshness_minutes=5, completeness_pct=99.9, uniqueness_check=False, schema_drift_alert=True),
    DataLayer.CLEANED: LayerSLO(freshness_minutes=15, completeness_pct=99.99, uniqueness_check=True, schema_drift_alert=True),
    DataLayer.AGGREGATED: LayerSLO(freshness_minutes=60, completeness_pct=99.0, uniqueness_check=False, schema_drift_alert=False),
    DataLayer.SERVING: LayerSLO(freshness_minutes=5, completeness_pct=100.0, uniqueness_check=True, schema_drift_alert=False),
}

def enforce_layer_policy(layer: DataLayer, metrics: dict) -> list:
    """根据分层策略返回所有不达标项"""
    slo = LAYER_DEFAULTS[layer]
    violations = []
    if metrics.get("freshness_min") > slo.freshness_minutes:
        violations.append(f"Freshness {metrics['freshness_min']}m exceeds SLO {slo.freshness_minutes}m")
    if metrics.get("completeness_pct", 100) < slo.completeness_pct:
        violations.append(f"Completeness below {slo.completeness_pct}%")
    return violations

# quality_gate.py
import pandas as pd
from great_expectations.core import ExpectationSuite
from great_expectations.dataset import PandasDataset

def run_quality_gate(df: pd.DataFrame, suite_name: str = "default") -> dict:
    """在 DataOps Pipeline 中作为门禁步骤运行"""
    dataset = PandasDataset(df)
    
    suite = ExpectationSuite(expectation_suite_name=suite_name)
    
    # 自定义期望：订单金额必须大于 0
    result_amount = dataset.expect_column_values_to_be_between(
        column="amount", min_value=0, mostly=0.999
    )
    
    # 货币字段必须在枚举范围内
    result_currency = dataset.expect_column_values_to_be_in_set(
        column="currency", value_set=["CNY", "USD", "EUR", "JPY"]
    )
    
    # 时间戳字段不允许为空
    result_time = dataset.expect_column_values_to_not_be_null(column="event_time")
    
    all_passed = all([
        result_amount["success"],
        result_currency["success"],
        result_time["success"]
    ])
    
    return {
        "gate_passed": all_passed,
        "details": {
            "amount_check": result_amount,
            "currency_check": result_currency,
            "timestamp_check": result_time
        },
        "action": "PROCEED" if all_passed else "BLOCK_AND_ALERT"
    }

# 模拟测试数据
test_df = pd.DataFrame({
    "order_id": ["A001", "A002", "A003"],
    "amount": [199.0, 0.0, 299.5],
    "currency": ["CNY", "USD", "EUR"],
    "event_time": pd.to_datetime(["2026-09-01 10:00", "2026-09-01 10:05", "2026-09-01 10:10"])
})

print(run_quality_gate(test_df))

# lineage_collector.py
from typing import Set

class LineageGraph:
    def __init__(self):
        self.nodes = set()
        self.edges = []  # (source, target, edge_type)
    
    def add_transformation(self, job_id: str, inputs: Set[str], outputs: Set[str], sql_logic: str = ""):
        self.nodes.update(inputs)
        self.nodes.update(outputs)
        for i in inputs:
            for o in outputs:
                self.edges.append({
                    "source": i,
                    "target": o,
                    "job": job_id,
                    "logic_hash": hash(sql_logic) & 0xFFFFFFFF,
                    "type": "TRANSFORMATION"
                })
    
    def get_upstream(self, node: str, depth: int = 3) -> Set[str]:
        """逆向追溯上游血缘，用于变更影响分析"""
        visited = set()
        queue = [(node, 0)]
        while queue:
            current, d = queue.pop(0)
            if d >= depth:
                continue
            for edge in self.edges:
                if edge["target"] == current and edge["source"] not in visited:
                    visited.add(edge["source"])
                    queue.append((edge["source"], d + 1))
        return visited
    
    def get_downstream_impact(self, node: str) -> list:
        """正向分析影响面，用于发布前风险评估"""
        impacted = []
        for edge in self.edges:
            if edge["source"] == node:
                impacted.append(edge["target"])
        return impacted

# 示例：构建订单域的血缘图
graph = LineageGraph()
graph.add_transformation(
    job_id="etl_order_enrichment",
    inputs={"mysql.orders", "mysql.users"},
    outputs={"iceberg.order.events"},
    sql_logic="SELECT o.*, u.tier FROM orders o JOIN users u ON o.user_id = u.id"
)
graph.add_transformation(
    job_id="agg_daily_revenue",
    inputs={"iceberg.order.events"},
    outputs={"iceberg.order.daily_revenue"},
    sql_logic="SELECT dt, SUM(amount) FROM order.events GROUP BY dt"
)

print("Upstream of daily_revenue:", graph.get_upstream("iceberg.order.daily_revenue"))
print("Downstream impact of mysql.orders:", graph.get_downstream_impact("mysql.orders"))

# pipeline_definition.py
from dataclasses import dataclass
from typing import List

@dataclass
class PipelineStage:
    name: str
    image: str
    commands: List[str]
    artifacts: List[str]
    env: dict
    
@dataclass
class DataPipeline:
    name: str
    trigger: dict
    stages: List[PipelineStage]
    
    def to_ci_yaml(self) -> str:
        """生成面向 GitLab CI / GitHub Actions 的配置片段"""
        lines = [f"# Auto-generated CI for {self.name}", "stages:"]
        for s in self.stages:
            lines.append(f"  - {s.name}")
        lines.append("")
        for s in self.stages:
            lines.append(f"{s.name}:")
            lines.append(f"  image: {s.image}")
            lines.append(f"  stage: {s.name}")
            lines.append("  script:")
            for cmd in s.commands:
                lines.append(f"    - {cmd}")
            if s.artifacts:
                lines.append("  artifacts:")
                lines.append("    paths:")
                for a in s.artifacts:
                    lines.append(f"      - {a}")
        return "\n".join(lines)

# 定义订单实时流处理流水线
order_stream_pipeline = DataPipeline(
    name="order_stream_etl",
    trigger={"branch": ["main"], "paths": ["pipelines/order/**"]},
    stages=[
        PipelineStage(
            name="unit_test",
            image="python:3.11-slim",
            commands=["pip install -r requirements.txt", "pytest tests/ -q"],
            artifacts=["coverage.xml"],
            env={"PYTEST_CURRENT_TEST": "1"}
        ),
        PipelineStage(
            name="sql_compile",
            image="sqlmesh/sqlmesh:latest",
            commands=["sqlmesh plan --auto-apply"],
            artifacts=["manifest.json"],
            env={}
        ),
        PipelineStage(
            name="data_diff",
            image="datafold/data-diff:latest",
            commands=["data-diff dev.prod order.events --conf datadiff.toml"],
            artifacts=["diff_report.html"],
            env={"DATAFOLD_API_KEY": "$DATAFOLD_API_KEY"}
        ),
        PipelineStage(
            name="deploy_prod",
            image="apache/airflow:2.8.0",
            commands=["airflow dags trigger order_events_v2", "airflow dags pause order_events_v1"],
            artifacts=[],
            env={"AIRFLOW_HOME": "/opt/airflow"}
        )
    ]
)

print(order_stream_pipeline.to_ci_yaml())

# data_bluegreen_deploy.py
class DataBlueGreenDeployer:
    def __init__(self, table_prefix: str, catalog):
        self.table_prefix = table_prefix
        self.catalog = catalog
        self.blue_suffix = "_v_blue"
        self.green_suffix = "_v_green"
    
    def deploy_new_version(self, sql_transform: str, validation_query: str) -> str:
        """在绿区部署新版本，验证通过后执行切换"""
        green_table = f"{self.table_prefix}{self.green_suffix}"
        blue_table = f"{self.table_prefix}{self.blue_suffix}"
        
        # 第一步：向绿区写入新逻辑计算结果
        self.catalog.execute(f"CREATE OR REPLACE TABLE {green_table} AS {sql_transform}")
        
        # 第二步：运行验证查询（行数、关键指标差异阈值）
        validation_result = self.catalog.execute(validation_query)
        assert validation_result["row_delta_pct"] < 0.01, "Row count deviation too large"
        assert validation_result["revenue_delta_pct"] < 0.001, "Revenue metric deviation too large"
        
        # 第三步：原子切换别名
        self.catalog.execute(f"ALTER TABLE {blue_table} RENAME TO {self.table_prefix}_old")
        self.catalog.execute(f"ALTER TABLE {green_table} RENAME TO {blue_table}")
        self.catalog.execute(f"ALTER TABLE {self.table_prefix}_old RENAME TO {green_table}")  # 轮换
        
        return blue_table  # 当前活跃表
    
    def rollback(self):
        """发现异常后的秒级回滚"""
        blue_table = f"{self.table_prefix}{self.blue_suffix}"
        green_table = f"{self.table_prefix}{self.green_suffix}"
        self.catalog.execute(f"ALTER TABLE {blue_table} RENAME TO {self.table_prefix}_tmp")
        self.catalog.execute(f"ALTER TABLE {green_table} RENAME TO {blue_table}")
        self.catalog.execute(f"ALTER TABLE {self.table_prefix}_tmp RENAME TO {green_table}")

# 使用示例
deployer = DataBlueGreenDeployer("order.daily_revenue", catalog=None)
# deployer.deploy_new_version(sql_transform="SELECT ...", validation_query="SELECT ...")

#!/bin/bash
# ci_data_diff.sh
# 在 GitHub Actions / GitLab CI 中运行

set -euo pipefail

SOURCE_TABLE="dev_order.daily_revenue"
TARGET_TABLE="prod_order.daily_revenue"
PRIMARY_KEY="dt"
REPORT_DIR="./reports"

mkdir -p "$REPORT_DIR"

echo "Running data diff between $SOURCE_TABLE and $TARGET_TABLE..."

data-diff \
  "snowflake://$SNOW_USER:$SNOW_PWD@$SNOW_ACCOUNT/$SNOW_DB?warehouse=$SNOW_WH&role=$SNOW_ROLE" \
  "$SOURCE_TABLE" \
  "$TARGET_TABLE" \
  -k "$PRIMARY_KEY" \
  -w "dt >= CURRENT_DATE - 7" \
  --json-output > "$REPORT_DIR/diff.json"

# 解析 diff 结果，若差异行超过 0.1% 则阻断流水线
MISMATCH_PCT=$(jq '.summary.mismatch_percent' "$REPORT_DIR/diff.json")
THRESHOLD=0.1

if (( $(echo "$MISMATCH_PCT > $THRESHOLD" | bc -l) )); then
  echo "ERROR: Data diff mismatch ${MISMATCH_PCT}% exceeds threshold ${THRESHOLD}%"
  exit 1
fi

echo "Data diff passed. Mismatch: ${MISMATCH_PCT}%"

# finops_cost_allocator.py
from dataclasses import dataclass
from typing import List, Dict
from collections import defaultdict

@dataclass
class ResourceUsage:
    resource_id: str
    resource_type: str  # S3, EC2, EMR, SnowflakeCredit, BigQuerySlot
    cost_usd: float
    usage_hours: float
    tags: Dict[str, str]

@dataclass
class DataProductCost:
    dp_address: str
    total_usd: float
    breakdown: Dict[str, float]  # type -> cost
    efficiency_score: float  # 业务产出 / 成本

class FinOpsAllocator:
    def __init__(self):
        self.allocations = defaultdict(lambda: {"total": 0.0, "items": []})
    
    def ingest(self, usages: List[ResourceUsage]):
        """基于资源标签将成本归属到具体数据产品"""
        for u in usages:
            dp = u.tags.get("dataProduct", "unallocated")
            owner = u.tags.get("owner", "unknown")
            key = f"{dp}#{owner}"
            self.allocations[key]["total"] += u.cost_usd
            self.allocations[key]["items"].append(u)
    
    def get_dp_report(self) -> List[DataProductCost]:
        results = []
        for key, data in self.allocations.items():
            dp, _ = key.split("#")
            breakdown = defaultdict(float)
            for item in data["items"]:
                breakdown[item.resource_type] += item.cost_usd
            results.append(DataProductCost(
                dp_address=dp,
                total_usd=data["total"],
                breakdown=dict(breakdown),
                efficiency_score=0.0  # 需要关联业务指标计算
            ))
        return sorted(results, key=lambda x: x.total_usd, reverse=True)
    
    def detect_waste(self, threshold_idle_hours: int = 168) -> List[dict]:
        """识别长期空闲的高成本资源"""
        waste = []
        for key, data in self.allocations.items():
            for item in data["items"]:
                if item.resource_type in ("EMR", "DatabricksCluster") and item.usage_hours < 1:
                    waste.append({
                        "resource": item.resource_id,
                        "dp": key.split("#")[0],
                        "cost": item.cost_usd,
                        "recommendation": "Terminate or downsize idle cluster"
                    })
        return waste

# 使用示例
usages = [
    ResourceUsage("s3://lake/order/raw", "S3", 120.5, 720, {"dataProduct": "order_events", "owner": "order-squad"}),
    ResourceUsage("emr-001", "EMR", 450.0, 0.5, {"dataProduct": "order_events", "owner": "order-squad"}),
    ResourceUsage("snowflake_wh_001", "SnowflakeCredit", 800.0, 720, {"dataProduct": "finance_reports", "owner": "finance-squad"}),
]

allocator = FinOpsAllocator()
allocator.ingest(usages)
for r in allocator.get_dp_report():
    print(f"{r.dp_address}: ${r.total_usd:.2f} -> {r.breakdown}")
print("Waste:", allocator.detect_waste())

# auto_scaler.py
import statistics
from dataclasses import dataclass

@dataclass
class JobHistory:
    job_name: str
    input_records: int
    duration_min: float
    executor_count: int
    memory_gb: int

class SparkAutoScaler:
    def __init__(self, history: list):
        self.history = history
    
    def predict_resources(self, expected_records: int) -> dict:
        """基于线性回归思想，根据历史数据预测最优资源配置"""
        # 简化版：按 input_records / throughput 估算
        throughputs = [h.input_records / h.duration_min for h in self.history if h.duration_min > 0]
        avg_throughput = statistics.mean(throughputs) if throughputs else 100000
        
        estimated_min = max(expected_records / avg_throughput, 1)
        
        # 推荐 executor = max(2, 预期耗时目标下的并行度)
        target_duration_min = 10
        recommended_executors = max(2, int(estimated_min / target_duration_min * 2))
        
        # 推荐内存：基于历史比例
        mem_per_executor = int(statistics.median([h.memory_gb / h.executor_count for h in self.history]))
        if mem_per_executor < 4:
            mem_per_executor = 4
        
        return {
            "expected_duration_min": round(estimated_min, 2),
            "recommended_executors": recommended_executors,
            "memory_per_executor_gb": mem_per_executor,
            "total_memory_gb": recommended_executors * mem_per_executor,
            "spot_enabled": estimated_min > 30  # 长作业启用 Spot 实例
        }

history = [
    JobHistory("order_etl", 1_000_000, 12, 4, 32),
    JobHistory("order_etl", 5_000_000, 45, 8, 64),
    JobHistory("order_etl", 10_000_000, 78, 12, 96),
]
scaler = SparkAutoScaler(history)
print(scaler.predict_resources(expected_records=8_000_000))

# storage_lifecycle_policy.yaml
policies:
  - name: hot_to_warm
    selector:
      layer: raw
      age_days: ">= 7"
      last_access_days: ">= 3"
    action:
      type: transition
      target_storage: s3_standard_ia
      
  - name: warm_to_cold
    selector:
      layer: raw
      age_days: ">= 30"
      query_frequency_30d: "< 2"
    action:
      type: transition
      target_storage: glacier_instant
      
  - name: archive_aggregated
    selector:
      layer: aggregated
      age_days: ">= 90"
    action:
      type: transition
      target_storage: glacier_deep_archive
      
  - name: delete_temp
    selector:
      path_prefix: "/tmp/"
      age_days: ">= 1"
    action:
      type: delete
      approved: true
      notify_owner: true

# idp_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid

app = FastAPI(title="Data Platform IDP")

class CreateDataProductRequest(BaseModel):
    name: str
    domain: str
    owner: str
    input_source: str
    transformation_sql: Optional[str] = None
    output_format: str = "iceberg"
    schedule: str = "0 2 * * *"  # 默认每日凌晨 2 点
    budget_usd_monthly: float = 500.0

class DataProductResponse(BaseModel):
    id: str
    address: str
    status: str
    deployed_resources: list

@app.post("/v1/dataproducts", response_model=DataProductResponse)
def create_data_product(req: CreateDataProductRequest):
    """领域团队自助创建数据产品的入口"""
    if not req.name.startswith(req.domain):
        raise HTTPException(status_code=400, detail="Data product name must start with domain prefix")
    
    dp_id = str(uuid.uuid4())[:8]
    address = f"dataproduct://{req.domain}/{req.name}@1.0.0"
    
    # 1. 注册 Catalog 表
    # 2. 创建 Airflow DAG / Flink Job 占位符
    # 3. 绑定成本标签
    # 4. 初始化质量监控
    
    return DataProductResponse(
        id=dp_id,
        address=address,
        status="PROVISIONING",
        deployed_resources=[
            f"iceberg://lake/{req.domain}/{req.name}",
            f"airflow-dag://{req.domain}/{req.name}",
            f"monitoring://grafana/{req.domain}-{req.name}"
        ]
    )

@app.get("/v1/dataproducts/{dp_id}/cost")
def get_dp_cost(dp_id: str):
    """实时查询数据产品的资源消耗与预算使用率"""
    return {
        "dp_id": dp_id,
        "month_to_date_usd": 312.40,
        "budget_usd": 500.0,
        "utilization_pct": 62.5,
        "top_cost_drivers": [
            {"resource": "Snowflake_Warehouse", "cost_usd": 180.0},
            {"resource": "S3_Storage", "cost_usd": 80.4},
            {"resource": "Airflow_Tasks", "cost_usd": 52.0}
        ]
    }

@app.post("/v1/dataproducts/{dp_id}/suspend")
def suspend_dp(dp_id: str):
    """紧急止损接口：暂停非关键数据产品以释放资源"""
    return {"dp_id": dp_id, "action": "SUSPENDED", "cost_saving_estimate_usd": 120.0}

# platform_observability.py
from prometheus_client import Gauge, Histogram, start_http_server
import time

class DataPlatformMetrics:
    def __init__(self):
        self.pipeline_latency = Histogram(
            "dataplatform_pipeline_latency_seconds",
            "End-to-end latency of data pipelines",
            ["domain", "dp_name", "layer"]
        )
        self.quality_score = Gauge(
            "dataplatform_data_quality_score",
            "Quality score (0-100) of data products",
            ["domain", "dp_name"]
        )
        self.cost_daily = Gauge(
            "dataplatform_daily_cost_usd",
            "Daily cost attribution per data product",
            ["domain", "dp_name", "resource_type"]
        )
        self.active_contracts = Gauge(
            "dataplatform_cross_domain_contracts",
            "Number of active cross-domain data contracts",
            ["provider_domain", "consumer_domain"]
        )
    
    def record_pipeline_run(self, domain: str, dp_name: str, layer: str, latency_sec: float):
        self.pipeline_latency.labels(domain=domain, dp_name=dp_name, layer=layer).observe(latency_sec)
    
    def update_quality(self, domain: str, dp_name: str, score: float):
        self.quality_score.labels(domain=domain, dp_name=dp_name).set(score)

# 启动 metrics 端点
if __name__ == "__main__":
    metrics = DataPlatformMetrics()
    start_http_server(9090)
    while True:
        metrics.record_pipeline_run("order", "order_events", "raw", 4.5)
        metrics.update_quality("order", "order_events", 94.2)
        time.sleep(60)

# chaos_experiments.py
import random
import requests

class DataPlatformChaos:
    def __init__(self, api_base: str):
        self.api_base = api_base
    
    def simulate_schema_drift(self, table: str, add_column: bool = True):
        """模拟上游 schema 突然变更，验证数据产品韧性"""
        print(f"[CHAOS] Injecting schema drift into {table}")
        # 实际实现应调用底层元数据服务或执行 ALTER TABLE
        return {"experiment": "schema_drift", "target": table, "injected": True}
    
    def kill_random_executor(self, job_name: str):
        """模拟 Spark / Flink executor 故障，验证容错与重算"""
        print(f"[CHAOS] Killing random executor for job {job_name}")
        # 调用 YARN / K8s API 杀死 Pod
        return {"experiment": "executor_failure", "target": job_name}
    
    def network_partition_catalog(self, duration_sec: int = 30):
        """模拟元数据服务网络分区，验证离线计算能力"""
        print(f"[CHAOS] Partitioning catalog for {duration_sec}s")
        # 利用 Linux tc 命令或 Istio fault injection
        return {"experiment": "network_partition", "duration": duration_sec}
    
    def run_safety_check(self) -> bool:
        """演练前安全检查，避免在生产高峰窗口执行"""
        resp = requests.get(f"{self.api_base}/v1/platform/health")
        data = resp.json()
        if data.get("active_critical_pipelines", 0) > 3:
            print("SAFETY CHECK FAILED: Too many critical pipelines running")
            return False
        return True