数据中台项目深度拆解

面向后端工程师的教学文档。目标：读完能理解这个千万级数据中台的每个设计决策背后的"为什么"，并能将这些模式迁移到自己的项目中。

项目路径：data-center（数据中台：接入 → 标准化 → 供数）

项目解决什么问题
技术栈知识铺垫
总体架构与分层思想
接入层：三条链路的设计
存储层：TimescaleDB 与标准事件模型
调度层：Airflow 聚合、对账与自愈
查询层：四层抽象与防护
查询 Agent：自然语言到受控查询
控制面与多租户权限
可靠性设计模式总结（可迁移知识）
如果让你从零实现：演进路线

1. 项目解决什么问题

业务数据分散的困境：几十上百个区服，每个区服的业务服务器往本地磁盘写 TLog 文本文件（log_LoginLog_1.2026-01-28.1700，SOH \x01 分隔）；同时还有生产 Doris/DW。下游（BI、客服、运营）想查数据，但：

上游格式脏：不同业务字段名不同、分隔符不同、Schema 会演化
数据会重复：网络重试、补数重跑、文件重传都会产生重复事件
数据会断：中心宕机、业务服务器网络抖动
权限要隔离：A 项目的人不能看 B 项目的数据

数据中台的定位是数据底座（不是 BI 产品）：把脏数据源在入口处消化掉，中心沉淀为标准事件层，下游只通过受控 API 拿干净数据。

核心设计哲学（贯穿全项目）：

边界洁癖——脏数据只在入口消化，下游永远只见标准层
悲观假设——假设上游会重复、会断、会脏、会延迟，每个假设都有机制兜底
元数据驱动——业务、区服、表、字段、指标、公式全部入库配置，接新业务不改代码

2. 技术栈知识铺垫

理解项目前，先补齐这些技术点。每个都标注了在项目中的应用位置。

2.1 TimescaleDB（时序 PostgreSQL 扩展）

数据中台的存储选型不是 ClickHouse/Doris，而是 TimescaleDB，原因：单组织单实例部署、需要完整的 PostgreSQL 生态（事务、JSONB、存储过程、advisory lock），且数据量在 PG 可承受范围内。

Hypertable（超表）：逻辑上是一张表，物理上按时间列自动切分为多个 chunk（本项目按天，见 database/migrations/016_create_events_table.sql:48-51）。

dc_events_dev_region_a (hypertable)
├── chunk_2026_07_18   ← 按 record_date 每天一个
├── chunk_2026_07_19
└── chunk_2026_07_20   ← 写入只落在最新 chunk（热数据）

为什么按天分区？三个收益：

查询裁剪：WHERE record_date >= ... 只扫描命中的 chunk，百万级表查询毫秒级
生命周期管理：删除 2 年前数据 = DROP 整个 chunk（秒级），而不是 DELETE 扫几亿行产生大量死元组和 vacuum 压力
压缩：老 chunk 整体转列存压缩（30 天后，016:81），压缩率通常 10x+

唯一约束的坑：TimescaleDB 要求唯一索引必须包含分区键，所以去重索引是 (client_uuid, record_date) 而不是单列 uuid（016:72-73）。这是"按 UUID 去重"在时序库上的标准解法。

压缩参数（016:76-80）：

timescaledb.compress_segmentby = 'event_name, source'   -- 同事件类型的行物理聚集
timescaledb.compress_orderby = 'record_date DESC, client_uuid'  -- 段内排序

segmentby 选低基数字段（事件类型只有 10 种），让同类型数据聚集在一起，压缩率最大化。

2.2 JSONB + 表达式索引

事件表用 data JSONB 存扩展字段（016:43），这是** schema-on-read** 的务实做法：

新事件类型加字段 → 不用 ALTER TABLE，不用发版
但高频查询字段（channel、platform）仍然建了表达式索引（016:66-69）：
```
CREATE INDEX ... ON t ((data->>'channel'), record_date)
```
这是"灵活"与"性能"的折中：JSONB 兜住变化，热字段提级为一等索引。

配套的是 safe_json_value() 系列数据库函数（015_add_safe_json_functions.sql）：历史数据里 data 可能是 jsonb 也可能是 text（脏数据），直接 data->>'field' 在边界情况会报错导致整个查询失败。safe 函数用 EXCEPTION WHEN OTHERS THEN RETURN NULL 兜底——让脏数据降级为 NULL，而不是让查询崩溃。

2.3 限流算法

项目里有两套限流（receiver/rate_limiter.py）：

算法	实现	场景
固定窗口计数器	Redis `INCR` + 过期时间	主路径（Redis 可用时）
令牌桶（Token Bucket）	进程内 `_LocalTokenBucket`（`rate_limiter.py:27-75`）	Redis 挂了时的 fallback

关键设计是 fail-open 降级链：Redis 连接失败 → 自动切本地令牌桶 → Redis 恢复后 30 秒重试自动切回（rate_limiter.py:81-120）。

为什么不 fail-close（Redis 挂了拒绝所有请求）？ 对数据接入服务来说，"收不到数据"比"限流失效"损失更大——限流失效最多 DB 压力大，数据丢了就真的丢了。这是典型的根据业务损失函数选择降级方向。

2.4 PostgreSQL Advisory Lock

Doris Pull Worker 需要防止同一 (game, region) 被两个进程并发拉取。用的是 PG 会话级咨询锁（receiver/workers/doris_checkpoint.py:25-70）：

lock_key = sha256(f"{namespace}:{game_id}:{region}")[:8] → int64
SELECT pg_try_advisory_lock(:lock_key)   # 拿不到立即返回 false，不阻塞
# ... 干完活 ...
SELECT pg_advisory_unlock(:lock_key)

为什么不用 Redis 分布式锁？DB 连接本来就在手里，零额外依赖；锁随会话结束自动释放，不存在"持锁进程崩溃导致死锁"问题；pg_try_ 非阻塞语义正好匹配"拿不到就跳过本轮"的调度模型。

2.5 幂等写入的三种武器

武器	项目中的位置	适用场景
`ON CONFLICT DO NOTHING`	事件入库（`event_routes.py:485-496`）	有唯一约束时的无脑重试安全写入
部分唯一索引 `WHERE client_uuid IS NOT NULL`	去重索引（`016:72`）	只对"有 uuid 的行"强制唯一，无 uuid 的行不受限
应用层预查重	批量查 `(uuid, record_date) IN (...)`（`event_routes.py:448-473`）	要在响应里告诉调用方"哪些 uuid 被跳过了"

注意 event_routes.py 是双管齐下：应用层先预查重（为了返回 skippedUuids 给上游做观测），DB 层再用 ON CONFLICT 兜底（防预查重到插入之间的并发间隙）。这是教科书级的防御纵深。

2.6 SQLite 作为边缘队列

Relay Agent 用 SQLite 做本地缓冲队列。选择理由：单文件、零依赖、事务保证、业务服务器上一定有。WAL 模式下读写并发够用。这是"边缘先存后发"模式的经典实现（类比 Kafka 的 page cache 思路，但简化到极致）。

3. 总体架构与分层思想

生产 Doris/DW → Doris Pull Worker ─┐
业务服务器 TLog ──→ Relay(SQLite缓冲) ──┼─→ Receiver 标准化 ─→ dc_events_{game}_{region}
Airflow/collector 兼容 Pull（补数） ────┘         │
                                       ┌───────┼─────────┐
                                     GA聚合表  用户事件查询  用户画像物化
                                              │
                              Receiver 查询 API（四层抽象）
                                              │
                        BI / 客服后台 / 运营 / 数据分析平台 / 查询 Agent

分层的本质问题是：上游表结构是下游不该知道的实现细节。

没有标准层：N 个上游表 × M 个下游 = N×M 个耦合点，上游改一个字段下游全炸
有标准层：N 个适配器 + M 个消费者，都只对标准层编程

这就是数仓理论里 ODS → DWD → DWS 的极简版：dc_tlog_*（ODS，已迁移淘汰）→ dc_events_*（DWD 标准事件层）→ ga_*（DWS 聚合层）。

单表 + event_name 的设计权衡（对比：每种事件一张表）：

维度	单表 + event_name（本项目）	每事件一表
新事件类型	零 DDL，直接写入	要建表、建 Cube schema
跨事件分析（漏斗）	单表自连接即可	多表 UNION
字段类型安全	弱（JSONB 里都是 text）	强（列类型）
单事件超大数据量	所有事件共享 chunk	可独立分区/清理

业务事件场景字段稀疏、类型演化快、需要频繁跨事件分析（登录→付费漏斗），单表胜出。聚合层的强类型需求由 ga_* 表承担。

4. 接入层：三条链路的设计

4.1 链路一：HTTP 推送 `POST /api/v1/events`

receiver/ingest/event_routes.py 是整个项目的"门面"，一个端点浓缩了接入层的全部防御手段。按请求处理顺序拆解：

请求进入
  │
  ├─ 1. API Key 鉴权 (Depends(verify_api_key))     ← 身份
  ├─ 2. Consumer 限流 (check_consumer_rate_limit)   ← 配额，429 + X-RateLimit-* 头
  ├─ 3. region 格式校验 (TABLE_NAME_RE)             ← 防注入：region 要拼进表名！
  ├─ 4. body 大小校验 ×2 (Content-Length + 实际 body) ← 防 1MB+ 超大 payload 攻击
  │        注意 event_routes.py:294-335 的双重校验：Content-Length 可伪造，必须读真实 body
  ├─ 5. 事件数量 ≤ 500 / 单事件 data ≤ 64KB / key ≤ 200 个  ← 资源防护
  ├─ 6. _ensure_table_exists：表不存在自动创建        ← 元数据驱动 + 小时级缓存防雪崩
  ├─ 7. clientTime 校验：±24h 外用 server_time 替代   ← 客户端时钟不可信
  ├─ 8. 分批插入：无 uuid 直接 executemany；
  │      有 uuid 的先请求内去重 → 批量预查重 → ON CONFLICT 兜底
  └─ 9. BackgroundTasks 异步记用量                    ← 不阻塞响应

值得背下来的细节：

为什么 game_id 从 API Key 取而不从请求体取（event_routes.py:283 verified_game_id = api_ctx.game_id）——身份即边界，请求体里的 game_id 永远不可信。A 业务的 Key 写不进 B 业务表。
clientTime ±24h 窗口（:222-230）：客户端时钟可能错、可能恶意。不拒绝（丢数据），而是修正为 server_time——数据可用性优先，时间字段标记不可信。
表存在性缓存每小时清空（:157-163）：防止内存无限增长，新业务接入最多延迟 1 小时被发现（可接受），不用引入 Redis。

4.2 链路二：Doris Pull Worker（生产主链路）

receiver/workers/ 下 13 个 doris_* 文件构成一个完整的可靠性子系统：

doris_pull_service.py      ← 主循环：按 checkpoint 增量拉取
doris_checkpoint.py        ← 进度持久化 + advisory lock 防并发
event_contract.py          ← 事件契约校验（见 4.4）
doris_event_writer.py      ← 写入标准事件层（复用同样的去重逻辑）
doris_dead_letter.py       ← 校验失败/写入失败 → DLQ
doris_pull_auto_repair.py  ← 检测缺口自动回补
doris_realtime_backfill.py ← 实时回补
doris_pull_reload.py       ← 人工触发重拉

Checkpoint 幂等模型：每个 (game_id, region) 在 dc_doris_checkpoint 表存一行 watermark（最后成功拉取的时间/ID）。每轮：

拿 advisory lock → 读 checkpoint → 拉 (watermark, now] 的数据
→ 校验 + 写入事件层 → 推进 checkpoint → 释放锁

崩溃恢复 = 重启后从 checkpoint 继续，重复拉取的部分由事件层的唯一索引去重吸收。"至少一次拉取 + 幂等写入 = 恰好一次语义"，这是流式数据工程的标准等式。

4.3 链路三：Relay Agent（先存后发）

relay/data_relay.py 是一个 1695 行的单文件程序，只依赖 requests——因为业务服务器环境不可控（没内网 PyPI、没 Docker、可能是 Windows Server）。

架构：

TLog 文件监听 ─┐
JSONL 监听    ─┼─→ 写入本地 SQLite（队列表）─→ Forwarder 线程
HTTP 代理 :17060 ┘                              │
                                                ├─ 每 5s 取 500 条
                                                ├─ POST /api/v1/events
                                                ├─ 成功 → 标记 sent
                                                └─ 失败 → 保留 pending，下轮重试

关键参数（data_relay.py:78-94）：BATCH_SIZE=500、FORWARD_INTERVAL=5s、REQUEST_TIMEOUT=30s、SQLite 队列保留 7 天（供 resend 命令）、本地归档 gzip 保留 90 天。

设计精髓：

中心宕机 = Relay 无事发生。事件在 SQLite 里堆着，中心恢复后自然追上。数据不丢失的保证被推到了离数据源最近的地方——这是对抗网络不可靠的最强姿势。
双保险：SQLite 队列 + JSONL 归档文件。即使 DB 损坏，还能从归档恢复。
可运维命令：resend --from DATE（重发）、export（导出）、purge（清理）、import-state（从旧版迁移状态）——生产工具必须自带"后悔药"。
FILE_QUIET_SECONDS=60：文件 60 秒没变化才认为写完，避免读到写了一半的 TLog 文件。

4.4 事件契约（Event Contract）——标准化的灵魂

receiver/workers/event_contract.py 定义了"什么是一个合法的标准事件"：

10 个标准事件（:28-39）：login / register / create_role / activate / online / charge / refund / currency_produce / currency_consume / item_change
每类事件的必填 data 字段（:53-64）：如 charge 必须有 amount/currency/order_id/is_first_pay
别名映射（:46-48）：payment → charge，上游历史叫法自动归一
字段路径映射 source_paths：上游叫 pay_amount，标准层叫 amount，契约里配映射
未知事件策略 unknown_event_strategy="quarantine"：不在契约里的事件不拒绝、不放入主表，隔离观察——既不丢数据，也不污染标准层

契约的存储有一个值得注意的演进（:19-27 注释）：早期硬编码常量 → 现在数据库表 dc_event_contract 是权威来源，常量只是新部署环境的 fallback。契约即配置，60 秒 TTL 缓存（:67）。

4.5 三条链路的角色边界

链路	角色	为什么存在
Doris Pull	生产主链路	数仓已有清洗后数据，拉过来标准化
Relay 推送	业务服务器实时链路	业务服务器改造成本最低：继续写 TLog 文件即可
Collector Pull	兼容/补数	历史回灌、Relay 部署前的过渡、排障时人工拉文件验证

三条链路汇入同一个标准事件层，写入逻辑（去重、契约、表结构）收敛到一处——链路边界清晰，收口统一。

5. 存储层：TimescaleDB 与标准事件模型

5.1 事件表结构（`016_create_events_table.sql`）

CREATE TABLE dc_events_{game}_{region} (
    id BIGSERIAL,
    record_date TIMESTAMPTZ NOT NULL,     -- 业务时间（分区键）
    ingest_time TIMESTAMPTZ DEFAULT NOW(),-- 到达时间（拉取游标用）
    source TEXT NOT NULL,                 -- 数据来源链路
    event_name TEXT NOT NULL,             -- 标准事件名
    zone_id INT, role_id BIGINT,
    account_id TEXT, device_id TEXT, session_id TEXT,
    client_ip TEXT, client_uuid TEXT,     -- 去重键
    data JSONB,                           -- 扩展字段
    server_time TIMESTAMPTZ DEFAULT NOW()
)

三个时间字段的分工值得细品：

record_date：业务发生时间，分区键 + 聚合维度
ingest_time：进入中心的时间，(ingest_time, client_uuid) 索引（016:62-63）支持"增量拉取游标"——下游同步器按 ingest_time 顺序消费
server_time：入库时间，审计用

业务时间和入库时间分离，才能处理"延迟到达的数据"（Relay 缓冲 30 天的数据到达时 record_date 是一个月前，但 ingest_time 是现在）——补数不破坏分区，回溯查询和增量同步各用各的时间。

5.2 动态建表：元数据驱动的物理实现

create_events_table(game_id, region) 是个 plpgsql 函数（016:11-88），Receiver 检测到新业务自动调用。一个函数完成：建表 → hypertable 化 → 8 个索引 → 压缩策略 → 保留策略。

对比"新业务接入要 DBA 手工建表"的传统流程，这是把数据库 DDL 也变成了元数据驱动的自动化。业务在 Admin 后台一点接入，物理表自动就绪。

5.3 生命周期策略

策略	配置	含义
压缩	30 天后（`016:81`）	热数据行存（写入快）→ 冷数据列存压缩（省 10x 空间）
保留	730 天（`016:84`，后由 019 调整）	到期自动 DROP chunk

早期模板（004）还演示了二维分区：时间按天 + 空间按 zone_id 30 个分区（支撑 300 区服）。空间分区让"查单个区服"也走分区裁剪。事件表最终没用空间分区（单业务单 region 一表已经天然隔离了业务维度）——分区维度要匹配真实查询模式，不是越多越好。

5.4 元数据表体系

dc_* 前缀表构成控制面的数据模型：

表	作用
`dc_game_config` / `dc_project`	业务归属项目（权限隔离的单位）
`dc_server_config` / `dc_zone_config`	物理服务器（SSH）/ 区服（300 条）
`dc_tlog_table_config`	字段映射 `field_mapping` JSONB：标准字段名 → 原始字段名
`dc_tlog_schema`	从 XML 解析的动态 Schema
`dc_collection_checkpoint` / `dc_doris_checkpoint`	两条链路的采集位点
`dc_event_contract` / `dc_event_contract_field`	事件契约（4.4 节）
`dc_admin_users` + role + project membership	管理后台权限
`dc_api_keys` / `dc_api_key_usage`	Key 鉴权 / 用量计量
`dc_external_table`	外部 SQL 查询的表白名单
`dc_schema_family` / `dc_schema_family_field`	逻辑字段 → 物理字段翻译

全项目没有一处硬编码的表名、字段名、业务列表。这就是为什么它号称"通用数据中台"而不是"某业务的数据管道"。

6. 调度层：Airflow 聚合、对账与自愈

约 25 个 DAG/模块（airflow/dags/），按职能分四组：

6.1 GA 日级聚合（`ga_daily_aggregation_dag.py`）

任务流：load（覆盖度检查）→ aggregate（聚合 SQL）→ report（物化状态上报）。

注意代码组织：DAG 文件只放编排，SQL 构造全部在 ga_daily_aggregation_sql.py 的 AGGREGATIONS 注册表里，字段映射走 dc_schema_family 元数据、查不到回落内置 FIELD_MAP（:64-90）。编排与计算分离——DAG 是流程图，不是 SQL 垃圾桶。

工程细节：retries=2、retry_delay=10min、execution_timeout=30min（default_args）——聚合任务允许失败重试，但必须有时限，不允许挂死占用 worker。

多版本并存：ga_daily_aggregation_dag（DW源）、_region_b（region_b 时区）、ga_doris_aggregation_dag（Doris 源）、ga_local_aggregation_dag（本地事件层源）——同一套聚合逻辑适配不同数据源和时区，通过 TableSourcePolicy（ga_table_policy.py）声明每张聚合表从哪个源取数、空数据是否合法。

6.2 对账（`ga_events_reconciliation_dag.py`）

聚合层是事件层的派生物，派生就可能出错。对账 DAG 定期比较：

事件层重算的 DAU/收入  vs  聚合表里的 DAU/收入

不一致 → 告警/触发修复。聚合结果不可信，必须被验证——这是数据平台和普通 CRUD 项目的分水岭。CRUD 项目"写进去就是对的"，数据项目"算出来的必须能对回去"。

6.3 自愈闭环

ga_rolling_repair_dag.py：滚动窗口重算最近 N 天——晚到的数据（Relay 补发、Doris 延迟）会让已产出的聚合过期，滚动修复让聚合自动收敛到正确值
doris_auto_repair_service.py：检测拉取缺口自动回补
ga_events_coverage.py + ga_materialization_status.py：覆盖度（该到的区服数据到了吗）和物化状态（每个分区聚合成功了吗）入库，Admin 后台可视化

聚合的正确性模型：最终一致 + 持续收敛。不追求一次算对，追求"错了我能发现、发现了我能重算、重算是幂等的（upsert）"。

6.4 维护（`maintenance_dag.py`）

DDL 健康巡检、统计信息更新、死元组清理等——数据库也是要被运营的。

7. 查询层：四层抽象与防护

查询 API 按抽象级别分四层，越往上自由度越低、安全性越高、业务语义越强：

Layer 4  查询 Agent        自然语言        → 受控工具调用
Layer 3  统一查询引擎      指标名+维度      → metric_registry 查注册表生成 SQL
Layer 2  Free/Aggregate    结构化 JSON     → query_builder 拼 SQL
Layer 1  SQL Query         裸 SQL          → sql_validator 7 层校验后执行

7.1 SQL Validator：七层校验链（`external/sql_validator.py`）

对外开裸 SQL 是危险的，但客服/BI 场景又确实需要。解法不是禁止，而是把 SQL 关进笼子：

SELECT-only：正则黑名单（INSERT/UPDATE/DROP/EXEC...），UNION 注入单独拦截（:50-51，防 ) UNION SELECT 拼出越权数据）
表名白名单：从 dc_external_table 元数据加载，FROM/JOIN 后的每个表名都必须在白名单内（:88-100）——用户能查什么表是配置出来的，不是猜出来的
危险关键字拦截
recorddate 范围强制 ≤ 90 天：没写时间范围或范围超限 → 拒绝/收窄。防止一条 SELECT * 扫两年数据打爆 DB
LIMIT 自动注入：没写 LIMIT 就补一个上限
SparkSQL 方言兼容（DW是 Spark 引擎）
Schema Family 逻辑字段翻译：用户写逻辑字段名 pay_amount，实际查询翻译成物理字段名——用户永远不知道物理 schema

配套：语句超时 300s（防慢查询占住连接池）、异步导出 Worker（大结果集走导出而不是同步响应）。

7.2 L2 查询缓存（`cache/l2_query.py`）

key = dc:query:{sha256(完整查询字符串)}
value = JSON 结果, TTL 300s, 单值 ≤ 1MB

设计取舍：

TTL 只有 5 分钟：数据平台对新鲜度敏感，5 分钟是"BI 刷图表狂点"和"数据延迟"的折中。不主动失效，靠 TTL 自然过期——聚合表每天才变一次，但查询参数组合是无限的，主动失效的 key 枚举成本远高于短 TTL
1MB 上限：防止一个大结果集挤爆 Redis 内存
独立 Redis DB（REDIS_DB_L2_QUERY）：缓存和限流分库，FLUSHDB 互不影响

7.3 统一查询引擎（`unified_query_engine/`）

把"指标"变成一等公民：

metric_registry.py（external 下）：指标注册表——指标名 → 定义（聚合表、公式、维度允许集）
metric_query_protocol.py：指标查询协议（请求长什么样）
metric_compute.py + realtime_expressions.py：派生指标计算（如 ARPU = 收入/DAU，公式存在元数据里）
ga_table_registry.py：自动发现 ga_* 聚合表
time_granularity.py：时间粒度归一（日/周/月）

为什么需要这一层：没有它，"DAU 怎么算"会散落在 20 个报表 SQL 里，口径漂移（运营算的 DAU 和客服算的 DAU 不一样）是数据平台最典型的信任危机。指标注册表 = 口径的唯一事实来源。

7.4 用户事件查询（`player_event_engine.py`）

客服场景专用："这个用户最近干了什么"。事件目录、明细、行为窗口、分群过滤——走 (account_id, record_date) 索引（016:56-57）。这是从"统计视角"到"个体视角"的补全。

8. 查询 Agent：自然语言到受控查询

receiver/agent/ 是给客服/分析用的 LLM 查询网关。核心设计决策：不让 LLM 生成 SQL，而是让 LLM 选择并填充受控工具。

用户问题 → LLM（workflow.py 编排）
   ├─ entity_resolver.py    "用户张三" → account_id
   ├─ policy.py             这个问题允许问吗？（scope 校验）
   └─ tools/                受控工具集：
        ├─ metric_tool.py            → 走统一查询引擎
        ├─ player_event_tool.py      → 走用户事件引擎
        └─ distribution_tool.py      → 分布查询
   → OutputGuardrail（guardrails.py）→ 答案评估（answer_eval.py）→ 响应

为什么不让 LLM 直接写 SQL：

安全：LLM 生成的 SQL 可能越权、可能是注入载体。工具调用把自由度从"任意 SQL"收窄到"有限函数 + 参数校验"，权限在工具层强制
正确性：LLM 不知道"DAU 在本平台的口径"，但 metric_tool 背后是指标注册表
可观测：每次调用留下 trace（trace.py）、答案被自动评估（answer_eval.py）

OutputGuardrail（guardrails.py）是输出侧护栏：剥离敏感字段（api_key、account_id、client_ip、device_id、raw_data——:9-17），剥离了就标记 quality.degraded = true；答案评估目前是 observe_only（只记录不拦截）——新安全机制先观察后执法，这是成熟的灰度思维。

9. 控制面与多租户权限

9.1 权限模型

部署形态：组织间物理隔离（单实例单组织），实例内逻辑隔离。权限分两类主体：

管理后台用户：dc_admin_users + role + project membership。角色是权限点集合（control_plane/access_control.py:8-60）：

project_owner:  project.* / game.* / api_key.* / dataset.* / tlog.* / push_config.*（全集）
project_admin:  同上但去掉 project.update/delete/manage_members
project_viewer: 只有 *.read

API 消费者：API Key 绑定 game + scope（事件推送 Key 不能调查询 API）+ Consumer Profile（限流档位）+ IP 白名单。

五级数据边界：Project → Game → Region → Dataset → API Key。每一级都有对应的校验点。

9.2 凭证与安全细节

credential_crypto.py / source_credentials.py：外部数据源（Doris/DW）的连接凭证加密落库，不明文
password_challenge.py：敏感操作（如查看 API Key 明文 api_key.reveal）需要重新输入密码——会话内二次确认，防"离开工位被操作"
ip_allowlist.py：查询类 Key 可绑 IP 白名单
secure_transport.py：传输层保护

9.3 用量追踪（`usage_tracking.py`）

写路径设计：Redis 实时 INCR（每次请求 O(1)，不落库）→ 每 5 分钟定时批量落 dc_api_key_usage。高频计数绝不同步写 DB——和另一个同类项目的限流计数同理，是"热路径异步化"的标准实践。配合事件入库时的 BackgroundTasks.add_task(record_usage, ...)（event_routes.py:512），响应返回后才记账。

9.4 监控（`monitoring/`）

Prometheus + Grafana + Alertmanager + exporters，shared/metrics.py 暴露应用指标，shared/alerts.py 告警。覆盖：接入速率、DLQ 深度、checkpoint 滞后、Doris Pull 健康、上线预检。

10. 可靠性设计模式总结（可迁移知识）

把全项目的防御手段抽象成模式清单——这些是可以搬进任何后端项目的：

#	模式	项目实例	一句话原则
1	至少一次 + 幂等 = 恰好一次	checkpoint 拉取 + 唯一索引去重	不要追求传输不重复，追求写入可重放
2	先存后发（Store-and-Forward）	Relay SQLite 缓冲 30 天	把可靠性推到离数据源最近的地方
3	死信队列	DLQ parquet 落盘 + 重放脚本	坏消息要存起来，不要扔
4	对账	聚合 vs 事件层核对 DAG	派生数据必须能对回源数据
5	滚动修复	重算最近 N 天聚合	最终一致靠持续收敛，不靠一次算对
6	防御纵深	body 大小查两次；去重查两次（预查 + ON CONFLICT）	单层防线一定有洞，关键路径叠两层
7	fail-open 降级	Redis 挂 → 本地令牌桶	按业务损失函数选降级方向
8	契约隔离	未知事件 quarantine 而非拒绝/放入	不丢数据，也不污染主流程
9	元数据驱动	dc_* 表配置一切，代码零硬编码	接新业务 = 插配置，不是改代码
10	悲观校验输入	clientTime ±24h 修正、Content-Length 不信	客户端的一切都可伪造
11	热路径异步化	用量 Redis 计数 5min 落库、BackgroundTasks	同步路径上只做必须同步的事
12	灰度执法	answer_eval observe_only	新安全策略先观察，确认不误伤再拦截

11. 如果让你从零实现：演进路线

这个项目 273 个迁移文件背后是一条典型的演进路径，值得作为自己项目的路线图：

阶段 1：能跑（MVP） 单链路（HTTP 推送）→ 单表事件 → 简单查询 API。此时没有契约、没有对账、没有权限。

阶段 2：不丢不重 加 checkpoint、唯一索引去重、Relay 先存后发、DLQ。数据可信是数据平台的生死线，优先于一切功能。

阶段 3：标准化 事件契约、字段映射、标准事件层取代直查 TLog。下游从"懂上游"变成"只懂标准层"。

阶段 4：供数产品化 聚合层 + 指标注册表 + 多层级查询 API + 权限/限流/用量计量。从"能查"变成"可运营"。

阶段 5：自愈与智能化 对账、滚动修复、覆盖度监控补齐闭环；查询 Agent 在自然语言入口上叠加受控查询能力。

每个阶段对应的原则：先保证数据不丢不重（正确性），再统一语义（标准化），再提升体验（产品化），最后才是智能（AI）。顺序反了——比如先接 BI 再做去重——就是灾难。

附：关键文件速查

主题	文件
事件接入全流程	`receiver/ingest/event_routes.py`
事件契约	`receiver/workers/event_contract.py`
Doris 拉取可靠性	`receiver/workers/doris_pull_service.py`, `doris_checkpoint.py`, `doris_dead_letter.py`
先存后发	`relay/data_relay.py`
事件表 DDL	`database/migrations/016_create_events_table.sql`
JSONB 安全函数	`database/migrations/015_add_safe_json_functions.sql`
SQL 防护	`receiver/external/sql_validator.py`
查询缓存	`receiver/cache/l2_query.py`
限流降级	`receiver/rate_limiter.py`
权限点定义	`receiver/control_plane/access_control.py`
Agent 护栏	`receiver/agent/guardrails.py`, `workflow.py`, `policy.py`
聚合编排	`airflow/dags/ga_daily_aggregation_dag.py` + `ga_daily_aggregation_sql.py`
对账/修复	`airflow/dags/ga_events_reconciliation_dag.py`, `ga_rolling_repair_dag.py`

目录​

1. 项目解决什么问题​

2. 技术栈知识铺垫​

2.1 TimescaleDB（时序 PostgreSQL 扩展）​

2.2 JSONB + 表达式索引​

2.3 限流算法​

2.4 PostgreSQL Advisory Lock​

2.5 幂等写入的三种武器​

2.6 SQLite 作为边缘队列​

3. 总体架构与分层思想​

4. 接入层：三条链路的设计​

4.1 链路一：HTTP 推送 POST /api/v1/events​

4.2 链路二：Doris Pull Worker（生产主链路）​

4.3 链路三：Relay Agent（先存后发）​

4.4 事件契约（Event Contract）——标准化的灵魂​

4.5 三条链路的角色边界​

5. 存储层：TimescaleDB 与标准事件模型​

5.1 事件表结构（016_create_events_table.sql）​

5.2 动态建表：元数据驱动的物理实现​

5.3 生命周期策略​

5.4 元数据表体系​

6. 调度层：Airflow 聚合、对账与自愈​

6.1 GA 日级聚合（ga_daily_aggregation_dag.py）​

6.2 对账（ga_events_reconciliation_dag.py）​

6.3 自愈闭环​

6.4 维护（maintenance_dag.py）​

7. 查询层：四层抽象与防护​

7.1 SQL Validator：七层校验链（external/sql_validator.py）​

7.2 L2 查询缓存（cache/l2_query.py）​

7.3 统一查询引擎（unified_query_engine/）​

7.4 用户事件查询（player_event_engine.py）​

8. 查询 Agent：自然语言到受控查询​

9. 控制面与多租户权限​

9.1 权限模型​

9.2 凭证与安全细节​

9.3 用量追踪（usage_tracking.py）​

9.4 监控（monitoring/）​

10. 可靠性设计模式总结（可迁移知识）​

11. 如果让你从零实现：演进路线​

附：关键文件速查​

目录