数据中台架构：

┌─────────────────────────────────────────────────┐
│                  数据服务层                      │
│  (API 服务、自助取数、数据产品)                   │
├─────────────────────────────────────────────────┤
│                  数据应用层                      │
│  (报表、分析、算法模型)                          │
├─────────────────────────────────────────────────┤
│                  数据资产层                      │
│  (指标体系、标签体系、主数据)                    │
├─────────────────────────────────────────────────┤
│                  数据开发层                      │
│  (ETL、数仓、实时计算)                          │
├─────────────────────────────────────────────────┤
│                  数据集成层                      │
│  (数据采集、同步、接入)                          │
├─────────────────────────────────────────────────┤
│                  基础设施层                      │
│  (存储、计算、调度)                             │
└─────────────────────────────────────────────────┘

项目结构示例：
data-platform/
├── dags/                  # 调度 DAG 定义
├── models/                # 数仓模型定义
│   ├── ods/
│   ├── dwd/
│   ├── dws/
│   └── ads/
├── scripts/               # ETL 脚本
├── tests/                 # 测试用例
├── docs/                  # 文档
└── infra/                 # 基础设施定义

# 数据质量测试示例
def test_order_data_quality():
    # 完整性测试
    null_ratio = execute_sql("""
        SELECT SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END) / COUNT(*)
        FROM dwd_order_detail WHERE dt = '${bizdate}'
    """)
    assert null_ratio < 0.01, f"order_id 空值率 {null_ratio} 超过阈值"

    # 一致性测试
    order_sum = execute_sql("SELECT SUM(amount) FROM dwd_order_detail WHERE dt = '${bizdate}'")
    payment_sum = execute_sql("SELECT SUM(amount) FROM dwd_payment_detail WHERE dt = '${bizdate}'")
    diff_ratio = abs(order_sum - payment_sum) / order_sum
    assert diff_ratio < 0.05, f"订单金额和支付金额差异 {diff_ratio} 超过阈值"

# GitLab CI 示例
stages:
  - lint
  - test
  - deploy

sql_lint:
  stage: lint
  script:
    - sqlfluff lint models/

unit_test:
  stage: test
  script:
    - pytest tests/unit/

integration_test:
  stage: test
  script:
    - pytest tests/integration/

deploy_to_prod:
  stage: deploy
  script:
    - dbt run --target prod
  only:
    - main

大数据成本构成：

存储成本（约 20-30%）
├── 热数据（频繁访问）
├── 温数据（偶尔访问）
└── 冷数据（几乎不访问）

计算成本（约 50-60%）
├── 固定资源（常驻集群）
├── 弹性资源（按需扩缩）
└── 临时资源（Spot/抢占式）

其他成本（约 10-20%）
├── 网络传输
├── 人力成本
└── 软件许可

-- 识别冷数据
SELECT
    table_name,
    MAX(query_time) as last_query_time,
    DATEDIFF(CURRENT_DATE, MAX(query_time)) as days_since_last_query
FROM table_access_log
GROUP BY table_name
HAVING days_since_last_query > 90;

-- 存储分级策略
-- 热数据：SSD，高可用
-- 温数据：HDD，标准存储
-- 冷数据：归档存储，按需恢复

资源利用率分析：
- 任务申请了 100G 内存，实际峰值只用了 20G → 缩减资源
- 任务每天跑，但数据一周才更新一次 → 调整调度频率
- 多个任务重复读取同一份数据 → 合并任务或缓存中间结果

# 按需扩缩容示例
def get_cluster_size(hour):
    """根据时段动态调整集群规模"""
    if 2 <= hour <= 8:  # 凌晨批处理高峰
        return 100
    elif 9 <= hour <= 18:  # 白天查询高峰
        return 50
    else:  # 低峰时段
        return 20

ROI 计算示例：

投入：
- 基础设施成本：200万/年
- 团队人力成本：500万/年
- 总投入：700万/年

产出：
- 支撑的业务决策带来的收益：难以量化，但可以举例
- 节省的人工取数成本：假设原来 10 人取数，现在自助化，节省 100万/年
- 数据驱动的增长：A/B 测试优化带来 GMV 提升 5%，假设 GMV 10亿，增量 5000万

ROI = (产出 - 投入) / 投入

技术选型评估矩阵：

             重要性
              高
              │
     稳定性 ──┼── 先进性
              │
             低

左上：核心系统，选成熟稳定的技术
右上：战略投入，选有发展前景的技术
左下：边缘系统，选成本最低的方案
右下：实验项目，选最新最酷的技术

场景：需要建设实时数据平台

Spark Streaming:
+ 团队已经熟悉 Spark
+ 批流统一编程模型
+ 生态成熟
- 实时性不如 Flink（微批）
- 状态管理能力较弱

Flink:
+ 真正的流处理，延迟更低
+ 强大的状态管理
+ Exactly-Once 语义支持好
- 团队需要学习新技术
- 生态相对较新

决策：
- 如果延迟要求不高（分钟级），且团队 Spark 经验丰富 → Spark Streaming
- 如果延迟要求高（秒级），且愿意投入学习成本 → Flink
- 如果不确定，可以先用 Flink SQL 入门，降低学习成本

技术管理者的时间分配：

代码评审 / 架构设计：30%
  - 保持技术敏感度
  - 把控技术方向

团队管理：30%
  - 1:1 沟通
  - 绩效评估
  - 招聘面试

跨团队协作：25%
  - 项目协调
  - 资源争取
  - 利益相关者管理

个人提升：15%
  - 学习新技术
  - 行业交流

数据团队的理想配置：

架构师/技术专家：10%
  - 把控技术方向
  - 解决疑难问题

高级工程师：30%
  - 核心模块开发
  - 带领小团队

中级工程师：40%
  - 日常开发主力
  - 独立交付能力

初级工程师：20%
  - 学习成长
  - 处理简单任务

传统数据平台 vs AI 时代数据平台：

传统数据平台：
数据采集 → ETL → 数仓 → BI/报表 → 人工分析

AI 时代数据平台：
数据采集 → ETL → 数仓 → 特征工程 → 模型训练 → 模型服务 → 智能应用
                    ↓
              向量化 → 向量库 → RAG/LLM 应用
                    ↓
              实时数据 → 在线特征 → 实时推理

# 特征平台核心能力示例
from feast import FeatureStore

store = FeatureStore(repo_path=".")

# 定义特征视图
user_features = FeatureView(
    name="user_features",
    entities=[user_entity],
    ttl=timedelta(days=1),
    features=[
        Feature(name="total_orders", dtype=Float32),
        Feature(name="avg_order_amount", dtype=Float32),
        Feature(name="days_since_last_order", dtype=Int32),
    ],
    source=user_source,
)

# 在线获取特征（毫秒级延迟）
features = store.get_online_features(
    features=["user_features:total_orders", "user_features:avg_order_amount"],
    entity_rows=[{"user_id": "12345"}]
)

# 离线获取特征（用于模型训练）
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=["user_features:total_orders", "user_features:avg_order_amount"]
)

开源方案：
- Feast：轻量级，易于上手，社区活跃
- Hopsworks：功能完整，自带 MLOps 能力
- Feathr（LinkedIn 开源）：企业级，和 Spark 集成好

云厂商方案：
- AWS SageMaker Feature Store
- Google Vertex AI Feature Store
- 阿里云 PAI 特征平台

# 向量数据库使用示例（以 Milvus 为例）
from pymilvus import connections, Collection

# 连接向量库
connections.connect("default", host="localhost", port="19530")

# 定义 Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
]
schema = CollectionSchema(fields)
collection = Collection("knowledge_base", schema)

# 插入向量
embeddings = openai.Embedding.create(input=texts, model="text-embedding-3-small")
collection.insert([ids, embeddings, texts])

# 相似度检索
results = collection.search(
    data=[query_embedding],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"nprobe": 10}},
    limit=5,
    output_fields=["text"]
)

# MLOps 流水线定义示例（Kubeflow Pipelines）
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  name: ml-pipeline
spec:
  templates:
    - name: data-prep
      container:
        image: data-prep:v1
        command: [python, prepare_data.py]

    - name: training
      container:
        image: training:v1
        command: [python, train.py]
        resources:
          limits:
            nvidia.com/gpu: 1

    - name: evaluation
      container:
        image: eval:v1
        command: [python, evaluate.py]

    - name: deployment
      container:
        image: deploy:v1
        command: [python, deploy_model.py]

实验管理：
├── 实验跟踪（MLflow、Weights & Biases）
├── 参数管理
├── 指标记录
└── 模型版本控制

模型注册：
├── 模型存储与版本化
├── 模型元数据管理
├── 模型血缘追踪
└── 模型审批流程

模型部署：
├── 批量推理（Spark MLlib、Ray）
├── 在线推理（TensorFlow Serving、Triton）
├── 边缘推理（ONNX、TensorRT）
└── A/B 测试与灰度发布

模型监控：
├── 数据漂移检测
├── 模型性能监控
├── 预测分布监控
└── 告警与自动回滚

LLMOps 核心关注点：

Prompt 工程：
├── Prompt 版本管理
├── Prompt 测试与评估
├── Prompt 模板库
└── Prompt 优化（CoT、Few-shot）

RAG 管道：
├── 文档处理与分块策略
├── Embedding 模型选型
├── 检索策略优化
├── 上下文注入与生成
└── 幻觉检测与缓解

成本控制：
├── Token 使用量监控
├── 缓存策略（语义缓存）
├── 模型选择（大模型 vs 小模型）
└── 批量处理优化

安全与合规：
├── 输入过滤（Prompt Injection 防护）
├── 输出审核
├── PII 脱敏
└── 审计日志

# 企业级 RAG 应用架构
class EnterpriseRAGSystem:
    def __init__(self):
        self.embedding_model = OpenAIEmbedding("text-embedding-3-small")
        self.vector_store = MilvusVectorStore(collection="knowledge")
        self.llm = AzureOpenAI(model="gpt-4o")
        self.cache = SemanticCache(threshold=0.95)

    def query(self, question: str, user_context: dict) -> str:
        # 1. 语义缓存检查
        cached = self.cache.get(question)
        if cached:
            return cached

        # 2. 权限过滤
        filter_expr = self._build_permission_filter(user_context)

        # 3. 向量检索
        docs = self.vector_store.search(
            query_embedding=self.embedding_model.embed(question),
            filter=filter_expr,
            top_k=5
        )

        # 4. 重排序
        docs = self.reranker.rerank(question, docs)

        # 5. 生成回答
        context = "\n".join([d.text for d in docs])
        response = self.llm.generate(
            prompt=f"基于以下信息回答问题：\n{context}\n\n问题：{question}"
        )

        # 6. 缓存结果
        self.cache.set(question, response)

        return response

传统数据团队技能栈：
SQL → ETL → 数仓建模 → 报表开发

AI 时代数据团队技能栈（新增）：
├── 特征工程与特征管理
├── 模型开发基础理解
├── 向量数据库与 Embedding
├── Prompt 工程与 LLM 应用
├── MLOps 工具链
└── AI 应用架构设计

评估 AI 工具投入产出比：

GitHub Copilot：
- 成本：$19/人/月
- 效率提升：预估 30-50%（因人而异）
- 适用场景：日常编码、测试用例生成

Cursor/Claude：
- 成本：$20-40/人/月
- 效率提升：复杂任务提升更明显
- 适用场景：代码理解、重构、文档生成

ChatGPT Team：
- 成本：$25/人/月
- 适用场景：文档写作、方案设计、问题排查

决策建议：
- 全员配置基础 AI 工具（Copilot 或类似）
- 核心开发人员配置高级工具（Cursor + Claude Pro）
- 建立 AI 使用最佳实践和培训体系