数据中台对外开放数据查询接口文档

接入信息
认证与权限
接口一：聚合指标查询（推荐首选）
接口二：自由查询
接口三：标准 SQL 查询
接口四：冷查询（异步）
接口五：异步导出
接口选择决策树
安全约束
限流机制
错误码汇总
SDK 中台表说明
各业务数据表
完整场景示例
附录

1. 接入信息

项目	值
基础 URL	`http://dc.example.com:17030`
认证方式	请求头 `X-API-Key: {API Key}`
认证类型	query 类型 API Key
所有接口前缀	`/api/v1/external`
返回格式	JSON

API 端点总览

#	方法	路径	说明	响应模式	适用场景
1	POST	`/aggregate`	聚合指标查询	同步	看板、实时大盘
2	POST	`/free-query`	自由查询	同步	明细下钻、探索分析
3	POST	`/sql-query`	标准 SQL 查询	同步	复杂分析、SQL 直接写
4	POST	`/query`	冷查询	异步（轮询）	批量报表、预定义数据集
5	GET	`/query/{job_id}`	轮询冷查询结果	同步	-
6	POST	`/query/{job_id}/cancel`	取消冷查询	同步	-
7	POST	`/export`	异步导出	异步（轮询）	大数据量导出 CSV/JSON
8	GET	`/export/{task_id}`	查询导出状态	同步	-

2. 认证与权限

2.1 认证方式

所有外部查询接口均在请求头中传入 X-API-Key：

X-API-Key: <DC_API_KEY>

这不同于 Cube.js 的 Authorization 头。Cube.js 使用 Authorization 头，查询 API 使用 X-API-Key 头。

2.2 Scope 权限模型

每个 query 类型 API Key 绑定一种 scope：

Scope 类型	scope_ref	权限范围	支持接口
`game`	game_id	仅限查询该业务	全部
`project`	project_id	可查询该项目下所有业务	全部
`dataset`	dataset_name	仅限查询指定数据集	仅冷查询

2.3 认证失败响应

// HTTP 401 - Key 无效
{ "detail": "Invalid or inactive query API Key" }

// HTTP 403 - 无权访问
{ "detail": "API Key 无权查询 game_id='xxx'" }

3. 接口一：聚合指标查询

POST /api/v1/external/aggregate

最推荐的首选接口。查 DAU、付费金额、付费人数、在线峰值等核心指标。当日或未最终物化日期从 dc_events_* 事件表实时/近实时计算，历史日期自动路由到 GA 物化日表。Redis 当前用于查询结果缓存、限流和异步结果缓存，不作为实时指标源。

请求参数

{
  "game_id": "game_a",
  "region": "region_a",
  "metrics": ["dau", "revenue", "pay_users", "pay_rate", "arpu", "online_peak"],
  "start_date": "2026-05-12",
  "end_date": "2026-05-18"
}

字段	类型	必填	说明
`game_id`	string	是	业务标识
`region`	string	否	区域，对外统一使用 `"region_a"`，默认 `"region_a"`
`metrics`	list[string]	是	指标列表，1-20 个
`start_date`	date	是	起始日期 YYYY-MM-DD
`end_date`	date	是	结束日期 YYYY-MM-DD

支持的全部指标

指标名	标签	类型	数据源	说明
`dau`	DAU	count	Events / GA	日活跃用户数
`new_users`	New Users	count	GA	新增用户数
`revenue`	Revenue	sum	Events / GA	充值总金额
`pay_users`	Pay Users	count	GA	付费用户数
`online_peak`	Online Peak	count	Events / GA	在线峰值
`pay_rate`	Pay Rate	computed	计算	付费率 = pay_users / dau
`arpu`	ARPU	computed	计算	人均收入 = revenue / dau
`arppu`	ARPPU	computed	计算	付费人均收入 = revenue / pay_users
`retention_d1`	D1 Retention	count	GA	次日留存率
`retention_d7`	D7 Retention	count	GA	7 日留存率
`ltv_d30`	LTV D30	computed	计算	30 天 LTV = arpu x 30

数据源自动路由

end_date >= 今天或物化未最终完成 -> dc_events_* 事件表实时/近实时计算
end_date < 今天且 GA 物化完成       -> GA TimescaleDB 物化日表

计算指标（pay_rate, arpu, arppu, ltv_d30）由引擎先查依赖指标再计算，无需手动处理。

响应示例

{
  "game_id": "game_a",
  "region": "region_a",
  "start_date": "2026-05-12",
  "end_date": "2026-05-18",
  "metrics": [
    { "metric": "dau", "value": 85230.0, "label": "DAU", "source": "ga" },
    { "metric": "revenue", "value": 1245600.5, "label": "Revenue", "source": "ga" },
    { "metric": "pay_users", "value": 3420.0, "label": "Pay Users", "source": "ga" },
    { "metric": "pay_rate", "value": 0.0401, "label": "Pay Rate", "source": "computed" },
    { "metric": "arpu", "value": 14.62, "label": "ARPU", "source": "computed" },
    { "metric": "online_peak", "value": 12100.0, "label": "Online Peak", "source": "ga" }
  ]
}

source 字段说明数据来自哪里："events" 表示事件事实层实时/近实时计算，"ga" 表示历史物化表，"computed" 表示引擎计算，"unavailable" 表示当前指标不可用。

cURL 示例

# 查询最近 7 天核心指标
curl -X POST 'http://dc.example.com:17030/api/v1/external/aggregate' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_a",
    "metrics": ["dau", "revenue", "pay_users", "pay_rate", "arpu"],
    "start_date": "2026-05-12",
    "end_date": "2026-05-18"
  }'

# 查询今日实时 DAU + 在线峰值
curl -X POST 'http://dc.example.com:17030/api/v1/external/aggregate' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_b",
    "metrics": ["dau", "online_peak", "revenue"],
    "start_date": "2026-05-19",
    "end_date": "2026-05-19"
  }'

Python 示例

import requests

BASE = "http://dc.example.com:17030/api/v1/external"
HEADERS = {"X-API-Key": "<DC_API_KEY>", "Content-Type": "application/json"}


def get_core_kpi(game_id: str, start: str, end: str) -> dict[str, float]:
    """获取核心 KPI 指标"""
    resp = requests.post(
        f"{BASE}/aggregate",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "metrics": ["dau", "revenue", "pay_users", "pay_rate", "arpu", "arppu"],
            "start_date": start,
            "end_date": end,
        },
    )
    resp.raise_for_status()
    data = resp.json()
    return {m["metric"]: m["value"] for m in data["metrics"]}


# 使用
kpis = get_core_kpi("game_a", "2026-05-12", "2026-05-18")
print(f"DAU: {kpis['dau']:.0f}, 收入: {kpis['revenue']:.2f}, 付费率: {kpis['pay_rate']:.2%}")

4. 接口二：自由查询

POST /api/v1/external/free-query

自由查询支持指标模式、公开 GA 表模式和原始表模式。指标模式走统一查询引擎，公开 GA 表模式走 GA table executor，原始表模式才会通过 DW SparkSQL 查询外部表。同步返回。

4.1 明细查询

返回原始数据行，适合查具体用户记录、道具流水、充值明细等。

请求结构：

{
  "game_id": "game_a",
  "table": "t_l_log_player_login",
  "columns": ["recorddate", "openid", "roleid", "areaid", "level", "ip"],
  "filters": {
    "record_date_start": "2026-05-18",
    "record_date_end": "2026-05-19",
    "areaid": "1"
  },
  "limit": 500
}

字段	类型	必填	默认值	说明
`game_id`	string	是	-	业务标识
`table`	string	是	-	表名
`columns`	list[string]	否	-	返回列。省略 = SELECT *
`filters`	dict	是	-	必须包含 `record_date_start` 和 `record_date_end`
`limit`	int	否	1000	1-10000
`cache_ttl`	int	否	3600	缓存秒数，0=不缓存

filters 说明：

record_date_start：必填，起始日期 YYYY-MM-DD
record_date_end：必填，结束日期（不包含，即 < 'end_date'）
其他 key：自定义维度过滤，均按 = 精确匹配

响应：

{
  "data": [
    { "recorddate": "2026-05-18", "openid": "oXXX...", "roleid": "123456", "areaid": "1", "level": "45", "ip": "example.com" },
    { "recorddate": "2026-05-18", "openid": "oYYY...", "roleid": "789012", "areaid": "1", "level": "32", "ip": "example.com" }
  ],
  "row_count": 2,
  "cached": false,
  "sql": null
}

4.2 聚合查询

指定 group_by + aggregates，进行 SUM/COUNT/AVG/MAX/MIN 聚合。

请求结构：

{
  "game_id": "game_a",
  "table": "t_l_log_player_login",
  "filters": {
    "record_date_start": "2026-05-12",
    "record_date_end": "2026-05-19"
  },
  "group_by": ["count_date"],
  "aggregates": {
    "dau": "COUNT(DISTINCT openid)",
    "login_times": "COUNT(*)",
    "avg_level": "AVG(level)"
  },
  "limit": 30
}

字段	类型	必填	说明
`group_by`	list[string]	是（聚合模式）	分组字段，支持 `DATE(col)`/`HOUR(col)` 等函数
`aggregates`	dict	是（聚合模式）	key=别名, value=`SUM(col)`/`COUNT(col)`/`AVG(col)`/`MAX(col)`/`MIN(col)`/`COUNT(*)`/`COUNT(DISTINCT col)`

group_by 支持的函数包装：

DATE(col) - 按天分组
HOUR(col) - 按小时分组
MONTH(col) - 按月分组
YEAR(col) - 按年分组
DAY(col) - 按日分组

响应：

{
  "data": [
    { "count_date": "2026-05-12", "dau": "82300", "login_times": "215000", "avg_level": "38.5" },
    { "count_date": "2026-05-13", "dau": "84500", "login_times": "221000", "avg_level": "38.7" }
  ],
  "row_count": 7,
  "cached": false,
  "sql": null
}

聚合查询返回的数值都是 string 类型（DW CSV 解析特性），调用方需自行 float() / int() 转换。

4.3 SDK 中台表自动隔离

查询 sdk_database_new 中的表时，系统自动注入 app_id 过滤，调用方无需传入。详见第 12 节。

4.4 cURL 示例

# 明细：查某天某区服登录记录
curl -X POST 'http://dc.example.com:17030/api/v1/external/free-query' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_a",
    "table": "t_l_log_player_login",
    "columns": ["count_date", "openid", "roleid", "areaid", "level"],
    "filters": {
      "record_date_start": "2026-05-18",
      "record_date_end": "2026-05-19",
      "areaid": "1"
    },
    "limit": 100
  }'

# 聚合：每日 DAU 趋势
curl -X POST 'http://dc.example.com:17030/api/v1/external/free-query' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_a",
    "table": "t_l_log_player_login",
    "filters": {
      "record_date_start": "2026-05-01",
      "record_date_end": "2026-05-19"
    },
    "group_by": ["count_date"],
    "aggregates": {
      "dau": "COUNT(DISTINCT openid)",
      "total_logins": "COUNT(*)"
    }
  }'

# SDK 中台表聚合：每日充值金额
curl -X POST 'http://dc.example.com:17030/api/v1/external/free-query' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_b",
    "table": "ads_sdk_charge_total_analyze",
    "filters": {
      "record_date_start": "2026-05-01",
      "record_date_end": "2026-05-19",
      "count_type": "1"
    },
    "group_by": ["count_date"],
    "aggregates": {
      "charge_amount": "SUM(charge_amount_cny)",
      "charge_users": "SUM(charge_account_cnt)"
    }
  }'

4.5 Python 封装示例

import requests
from typing import Optional

BASE = "http://dc.example.com:17030/api/v1/external"
HEADERS = {"X-API-Key": "<DC_API_KEY>", "Content-Type": "application/json"}


def free_query(
    game_id: str,
    table: str,
    start: str,
    end: str,
    *,
    columns: Optional[list[str]] = None,
    extra_filters: Optional[dict] = None,
    group_by: Optional[list[str]] = None,
    aggregates: Optional[dict[str, str]] = None,
    limit: int = 1000,
    cache_ttl: int = 3600,
) -> dict:
    """自由查询 - 明细 & 聚合统一入口"""
    filters = {"record_date_start": start, "record_date_end": end}
    if extra_filters:
        filters.update(extra_filters)

    payload: dict = {
        "game_id": game_id,
        "table": table,
        "filters": filters,
        "limit": limit,
        "cache_ttl": cache_ttl,
    }
    if columns:
        payload["columns"] = columns
    if group_by:
        payload["group_by"] = group_by
    if aggregates:
        payload["aggregates"] = aggregates

    resp = requests.post(f"{BASE}/free-query", headers=HEADERS, json=payload)
    resp.raise_for_status()
    return resp.json()


# ---- 用户活跃度示例 ----

# 示例 1: 查询每日 DAU + 新增登录次数（聚合模式）
result = free_query(
    "game_a", "t_l_log_player_login",
    "2026-05-12", "2026-05-19",
    group_by=["count_date"],
    aggregates={
        "dau": "COUNT(DISTINCT openid)",
        "login_times": "COUNT(*)",
    },
)
print("=== 每日活跃度 ===")
for row in result["data"]:
    print(f"  {row['count_date']}: DAU={row['dau']}, 登录次数={row['login_times']}")

# 示例 2: 查某天高等级活跃用户明细（明细模式）
result = free_query(
    "game_a", "t_l_log_player_login",
    "2026-05-18", "2026-05-19",
    columns=["openid", "roleid", "level", "areaid", "ip"],
    extra_filters={"areaid": "1"},
    limit=10,
)
print("\n=== 1区高活跃用户（最近登录）===")
for row in result["data"]:
    print(f"  roleid={row['roleid']}, level={row['level']}")

# 示例 3: 分时登录分布
result = free_query(
    "game_a", "t_l_log_player_login",
    "2026-05-18", "2026-05-19",
    group_by=["HOUR(recorddate)"],
    aggregates={
        "login_count": "COUNT(*)",
        "unique_users": "COUNT(DISTINCT openid)",
    },
)
print("\n=== 分时登录分布 ===")
for row in result["data"]:
    print(f"  时段{row['hour(recorddate)']}: 登录{row['login_count']}次, 独立用户{row['unique_users']}")

# 示例 4: 充值金额趋势
result = free_query(
    "game_a", "t_l_log_charge",
    "2026-05-01", "2026-05-19",
    group_by=["count_date"],
    aggregates={
        "total_money": "SUM(money)",
        "pay_orders": "COUNT(*)",
        "pay_users": "COUNT(DISTINCT openid)",
    },
)
print("\n=== 每日充值趋势 ===")
for row in result["data"]:
    print(f"  {row['count_date']}: 金额={row['total_money']}, 订单={row['pay_orders']}, 人数={row['pay_users']}")

5. 接口三：标准 SQL 查询

POST /api/v1/external/sql-query

直接写 SELECT 语句查询。SQL 引用指标虚拟表或公开 GA 表时走统一查询引擎/GA table executor；其他白名单外部表经过 7 层安全校验后通过 DW SparkSQL 执行。比 free-query 更灵活，支持 JOIN、子查询、复杂 WHERE 等。

请求参数

{
  "game_id": "game_a",
  "sql": "SELECT count_date, COUNT(DISTINCT openid) AS dau, COUNT(*) AS login_times FROM game_a_database.t_l_log_player_login WHERE recorddate >= '2026-05-12' AND recorddate < '2026-05-19' GROUP BY count_date ORDER BY count_date LIMIT 30",
  "use_cache": true
}

字段	类型	必填	默认值	说明
`game_id`	string	是	-	业务标识
`sql`	string	是	-	SELECT 语句，最大 32768 字符
`use_cache`	bool	否	true	是否使用 L2 缓存

7 层安全校验

层	校验内容	拦截示例
1	SELECT-only	INSERT/UPDATE/DELETE/DROP 等被拦截
2	表名白名单	只能查 `dc_external_table` 中注册的表
3	危险关键字	EXECUTE/CALL/GRANT/REVOKE 等被拦截
4	recorddate 范围 `<=` 90 天	WHERE 条件中日期范围强制限制
5	LIMIT 自动注入	无 LIMIT 自动补 `LIMIT 10000`
6	SparkSQL 方言兼容	适配 DW SparkSQL 特殊语法
7	Schema Family 翻译	逻辑字段名自动翻译为物理字段名

SQL 编写要点：

表名需写完整路径：{database}.{table_name}（系统会自动校验该表是否属于该 game）
时间过滤必须包含 recorddate >= 'YYYY-MM-DD' AND recorddate < 'YYYY-MM-DD'
SQL 必须是纯 SELECT 语句

响应

{
  "query_id": "sql_a1b2c3d4e5f6a7b8",
  "game_id": "game_a",
  "sanitized_sql": "SELECT count_date, COUNT(DISTINCT openid) AS dau ...",
  "rows": [
    ["2026-05-12", "82300", "215000"],
    ["2026-05-13", "84500", "221000"]
  ],
  "columns": ["count_date", "dau", "login_times"],
  "row_count": 7,
  "elapsed_ms": 1234.5,
  "source": "dw"
}

字段	说明
`query_id`	查询标识（SQL SHA256 前 16 位）
`sanitized_sql`	校验净化后的 SQL
`rows`	`list[list]` - 每行是值的列表
`columns`	列名列表，与 rows 顺序对应
`row_count`	结果行数
`elapsed_ms`	执行耗时（毫秒），缓存命中时为 0
`source`	`"cache"` 或 `"dw"`

与 free-query 不同，sql-query 返回 rows 是 list[list] 而非 list[dict]，需配合 columns 按索引取列。

cURL 示例

# 查询每日 DAU（SQL 模式）
curl -X POST 'http://dc.example.com:17030/api/v1/external/sql-query' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_a",
    "sql": "SELECT count_date, COUNT(DISTINCT openid) AS dau FROM game_a_database.t_l_log_player_login WHERE recorddate >= '\''2026-05-12'\'' AND recorddate < '\''2026-05-19'\'' GROUP BY count_date ORDER BY count_date LIMIT 30",
    "use_cache": true
  }'

# 查付费用户等级分布
curl -X POST 'http://dc.example.com:17030/api/v1/external/sql-query' \
  -H 'X-API-Key: <DC_API_KEY>' \
  -H 'Content-Type: application/json' \
  -d '{
    "game_id": "game_a",
    "sql": "SELECT level, COUNT(*) AS user_count, SUM(money) AS total_money FROM game_a_database.t_l_log_charge WHERE recorddate >= '\''2026-05-01'\'' AND recorddate < '\''2026-05-19'\'' GROUP BY level ORDER BY level LIMIT 100",
    "use_cache": false
  }'

Python 封装示例

def sql_query(
    game_id: str,
    sql: str,
    use_cache: bool = True,
) -> list[dict]:
    """标准 SQL 查询，返回 list[dict]"""
    resp = requests.post(
        f"{BASE}/sql-query",
        headers=HEADERS,
        json={"game_id": game_id, "sql": sql, "use_cache": use_cache},
    )
    resp.raise_for_status()
    result = resp.json()
    # 将 list[list] + columns 转换为 list[dict]
    return [dict(zip(result["columns"], row)) for row in result["rows"]]


# 示例：查询每日活跃 + 付费交叉分析
sql = """
SELECT
    a.count_date,
    a.dau,
    COALESCE(b.pay_users, 0) AS pay_users,
    COALESCE(b.total_revenue, 0) AS total_revenue,
    ROUND(COALESCE(b.pay_users, 0) * 100.0 / a.dau, 2) AS pay_rate
FROM (
    SELECT count_date, COUNT(DISTINCT openid) AS dau
    FROM game_a_database.t_l_log_player_login
    WHERE recorddate >= '2026-05-12' AND recorddate < '2026-05-19'
    GROUP BY count_date
) a
LEFT JOIN (
    SELECT count_date, COUNT(DISTINCT openid) AS pay_users, SUM(money) AS total_revenue
    FROM game_a_database.t_l_log_charge
    WHERE recorddate >= '2026-05-12' AND recorddate < '2026-05-19'
    GROUP BY count_date
) b ON a.count_date = b.count_date
ORDER BY a.count_date
LIMIT 30
"""

rows = sql_query("game_a", sql)
for r in rows:
    print(f"{r['count_date']}: DAU={r['dau']}, 付费人数={r['pay_users']}, "
          f"收入={r['total_revenue']}, 付费率={r['pay_rate']}%")

6. 接口四：冷查询（异步）

6.1 POST /api/v1/external/query - 提交冷查询

基于预定义数据集（dc_dataset_catalog）提交异步查询任务。适用于定期报表场景。

{
  "dataset_name": "game_a_login_summary",
  "filters": {
    "record_date_start": "2026-05-01",
    "record_date_end": "2026-05-19"
  },
  "columns": ["count_date", "dau", "login_times"],
  "limit": 1000
}

字段	类型	必填	说明
`dataset_name`	string	是	预定义数据集名称（管理员在后台发布）
`filters`	dict	是	必须包含 `record_date_start` 和 `record_date_end`
`columns`	list[string]	否	不传则返回全部字段
`limit`	int	否	默认 1000

响应 (202 Accepted)：

{
  "job_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "pending",
  "poll_url": "/api/v1/external/query/550e8400-e29b-41d4-a716-446655440000"
}

相同查询参数会命中 query_hash 去重，直接返回 "status": "completed"。

6.2 GET /api/v1/external/query/{job_id} - 轮询结果

GET /api/v1/external/query/{job_id}

响应：

{
  "status": "completed",
  "row_count": 7,
  "result_bytes": 2048,
  "data": [
    { "count_date": "2026-05-12", "dau": 82300, "login_times": 215000 }
  ],
  "cached": true,
  "created_at": "2026-05-19T10:30:00",
  "heartbeat_at": "2026-05-19T10:30:15",
  "error_message": null,
  "expires_at": "2026-05-19T11:30:00"
}

状态流转：

pending -> running -> completed / failed
                          v
                      cancelled (仅 pending/running 可取消)

6.3 POST /api/v1/external/query/{job_id}/cancel - 取消查询

取消处于 pending 或 running 的任务。

Python 冷查询封装

import time


def cold_query_and_wait(
    dataset_name: str,
    filters: dict,
    *,
    columns: Optional[list[str]] = None,
    limit: int = 1000,
    max_wait: float = 120.0,
) -> dict:
    """提交冷查询并等待结果"""
    # 1. 提交
    payload = {"dataset_name": dataset_name, "filters": filters, "limit": limit}
    if columns:
        payload["columns"] = columns

    resp = requests.post(f"{BASE}/query", headers=HEADERS, json=payload)
    resp.raise_for_status()
    job = resp.json()

    # 2. 轮询
    poll_url = f"{BASE}/query/{job['job_id']}"
    interval = 1.0
    elapsed = 0.0

    while elapsed < max_wait:
        if job["status"] in ("completed", "failed", "cancelled"):
            break

        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 1.5, 5.0)

        poll_resp = requests.get(poll_url, headers=HEADERS)
        poll_resp.raise_for_status()
        job = poll_resp.json()

    if job["status"] == "completed":
        return job
    elif job["status"] == "failed":
        raise RuntimeError(f"冷查询失败: {job.get('error_message')}")
    else:
        raise TimeoutError(f"冷查询超时 ({max_wait}s): status={job['status']}")


# 使用
result = cold_query_and_wait(
    "game_a_login_summary",
    {"record_date_start": "2026-05-01", "record_date_end": "2026-05-19"},
)
for row in result["data"]:
    print(row)

7. 接口五：异步导出

7.1 POST /api/v1/external/export - 提交导出任务

大数据量查询结果导出为 CSV/JSON 文件，上传到对象存储后返回下载 URL。适用于超过 10000 行的场景。

{
  "game_id": "game_a",
  "sql": "SELECT * FROM game_a_database.t_l_log_player_login WHERE recorddate >= '2026-05-01' AND recorddate < '2026-05-19'",
  "format": "csv",
  "filename": "login_20260501_0518"
}

字段	类型	必填	说明
`game_id`	string	是	业务标识
`sql`	string	是	SELECT 语句
`format`	string	否	`"csv"` 或 `"json"`，默认 csv
`filename`	string	否	不含后缀，为空则自动生成

响应：

{
  "task_id": "a1b2c3d4e5f6",
  "status": "pending",
  "message": "导出任务已提交，请通过 GET /api/v1/external/export/a1b2c3d4e5f6 查询进度"
}

7.2 GET /api/v1/external/export/{task_id} - 查询导出状态

GET /api/v1/external/export/{task_id}

响应（completed）：

{
  "task_id": "a1b2c3d4e5f6",
  "status": "completed",
  "download_url": "https://cloud.example.com/exports/login_20260501_0518.csv?sign=...",
  "expires_at": "2026-05-19T11:30:00",
  "rows_exported": 250000,
  "error_message": null,
  "created_at": "2026-05-19T10:30:00",
  "updated_at": "2026-05-19T10:35:00"
}

状态流程：

pending -> running -> completed (download_url 可获取)
                  -> failed (error_message 有值)

download_url 是对象存储预签名 URL，默认 1 小时有效。

Python 导出封装

def export_and_download(
    game_id: str,
    sql: str,
    output_path: str,
    *,
    format: str = "csv",
    filename: str = "",
    max_wait: float = 600.0,
) -> int:
    """提交导出任务，等待完成，下载到本地。返回导出行数。"""
    # 1. 提交
    resp = requests.post(
        f"{BASE}/export",
        headers=HEADERS,
        json={"game_id": game_id, "sql": sql, "format": format, "filename": filename},
    )
    resp.raise_for_status()
    task = resp.json()
    task_id = task["task_id"]

    # 2. 轮询
    interval = 3.0
    elapsed = 0.0
    while elapsed < max_wait:
        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 1.5, 15.0)

        resp = requests.get(f"{BASE}/export/{task_id}", headers=HEADERS)
        resp.raise_for_status()
        status = resp.json()

        if status["status"] == "completed":
            break
        elif status["status"] == "failed":
            raise RuntimeError(f"导出失败: {status.get('error_message')}")

    # 3. 下载
    download_url = status.get("download_url")
    if not download_url:
        raise RuntimeError("导出完成但无下载 URL")

    file_resp = requests.get(download_url)
    file_resp.raise_for_status()
    with open(output_path, "wb") as f:
        f.write(file_resp.content)

    print(f"已下载: {output_path} ({len(file_resp.content)} bytes, {status['rows_exported']} 行)")
    return status["rows_exported"]


# 导出最近一周所有登录日志
export_and_download(
    "game_a",
    "SELECT * FROM game_a_database.t_l_log_player_login WHERE recorddate >= '2026-05-12' AND recorddate < '2026-05-19'",
    "./login_export.csv",
    filename="login_weekly",
)

8. 接口选择决策树

需要查什么？
│
├─ 查核心指标（DAU/收入/付费率/ARPU/在线峰值/留存）
│   -> POST /aggregate  首选，自动路由实时/历史，计算指标自动算
│
├─ 查明细数据 / 自定义聚合（按表查、灵活组合）
│   ├─ 需求简单（单表、等值过滤、基础聚合）
│   │   -> POST /free-query  简单 JSON 参数，无需写 SQL
│   │
│   └─ 需求复杂（JOIN、子查询、CASE WHEN、多条件组合）
│       -> POST /sql-query   直接写 SQL，7层安全校验
│
├─ 基于预定义数据集定期出报表
│   -> POST /query  （冷查询，异步轮询）
│
└─ 导出大量数据（>10000 行）到文件
    -> POST /export  （异步导出 CSV/JSON，返回下载链接）

各接口对比一览

维度	aggregate	free-query	sql-query	冷查询	export
响应速度	`<500ms`	1-10s	1-10s	分钟后级	分钟后级
返回上限	20 指标	10000 行	100000 行	10000 行	无限制
数据源	Events+GA	UQE/GA/DW	UQE/GA/DW	DW	DW
缓存	Redis L2	Redis(可选)	Redis L2	Redis	无
实时数据	当日事件表	指标模式可支持	指标/GA 模式可支持	否	否
SQL 复杂度	无	无	JOIN/子查询	无	JOIN/子查询
认证	X-API-Key	X-API-Key	X-API-Key	X-API-Key	X-API-Key

9. 安全约束

9.1 SQL 注入防护

所有接口在 SQL 到达数据库之前都经过严格的白名单 + 正则校验：

输入类型	校验规则
表名/列名	`^[a-zA-Z_][a-zA-Z0-9_]*$`
日期	YYYY-MM-DD + `strptime` 验证
聚合表达式	`^(SUM
GROUP BY	纯列名或 `^(DATE
SQL 语句	仅 SELECT，表名白名单，危险关键字拦截
字符串值	单引号转义为 `''`

不合法输入直接返回 400，不会到达 SQL 层。

9.2 通用约束

约束项	值
时间范围上限	90 天
free-query limit	1 - 10000
sql-query 自动 LIMIT	10000（无 LIMIT 时自动注入）
DW 查询超时	120 秒
结果缓存上限	5 MB
cache_ttl 范围	0 - 86400
SQL 最大长度	32768 字符

10. 限流机制

每个 API Key 受 每分钟请求数 限制（默认 10 次），所有对外接口共用限额。

限流失败响应：

HTTP 429 Too Many Requests
X-RateLimit-Limit: 10
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1715500800

{ "detail": "查询请求过于频繁，请稍后重试" }

建议： 调用方应在客户端实现重试逻辑，收到 429 后等待 Retry-After 秒再重试。

11. 错误码汇总

状态码	场景	典型 detail
400	参数验证失败	`"filters 必须包含 record_date_start 和 record_date_end"`
400	参数验证失败	`"时间范围不能超过 90 天"`
400	参数验证失败	`"SQL 校验失败: ..."`
400	不支持的指标	`"不支持的指标: xxx，可用: dau, revenue, ..."`
401	API Key 无效	`"Invalid or inactive query API Key"`
403	无权访问	`"API Key 无权查询 game_id='xxx'"`
403	不支持的操作	`"dataset scope 的 API Key 不支持自由查询"`
404	资源不存在	`"数据集 'xxx' 不存在或未发布"`
404	任务不存在	`"查询任务不存在"`
404	表不存在	`"表 'xxx' 在该业务绑定的数据库中不存在"`
413	结果过大	`"查询结果 xx 字节超过上限"`
429	限流	`"查询请求过于频繁，请稍后重试"`
502	DW 失败	`"DW 查询失败: ..."`

12. SDK 中台表说明

sdk_database_new 是所有业务共享的 SDK 数据库，通过 app_id 隔离。查询时 系统自动注入 app_id 过滤。

常用 SDK 表

表名	说明	常用列
`ads_sdk_charge_total_analyze`	充值汇总	count_date, charge_amount_cny, charge_account_cnt, count_type
`ads_sdk_register_account_analyze_total`	注册分析汇总	count_date, register_account_cnt
`ads_sdk_createrole_account_retention`	创角留存	count_date, day1_retention_rate ~ day180_retention_rate
`ods_sdk_login`	登录原始日志	app_id, openid, event_time, ip, channel
`ods_sdk_charge_success`	充值成功日志	app_id, openid, order_amount, order_id
`dim_sdk_app_timezone`	时区映射	app_id, timezone

13. 各业务数据表

13.1 业务标识

业务	game_id	数据库
game_a	`game_a`	game_a_database
game_b	`game_b`	game_b_database
game_c	`game_c`	game_c_database
game_d	`game_d`	game_d_database

13.2 字段风格差异

字段含义	game_a/game_b 系列	game_c/game_d
区服ID	`areaid` (int)	`siteid` (int)
设备号	`deviceno`	`deviceid`
充值金额	`money` (double)	`chargemoney` (double)
在线人数	`online` (int)	`onlineplayer` (int)
渠道	`channelid`	`reg_channel`
时间分区	`recorddate` / `count_date`	`recorddate` / `count_date`

13.3 核心表速查

表名	用途	game_a/game_b	game_c/game_d
`t_l_log_player_login`	登录日志	是	是
`t_l_log_player_logout`	登出日志	是	是
`t_l_log_charge`	充值记录	是	是
`t_l_log_createuser`	创建账号	是	-
`t_l_log_createrole`	创建角色	是	是
`t_l_log_player_register`	注册日志	-	是
`t_l_log_reg_online`	在线人数	是	-
`t_l_log_regonline`	在线人数	-	是
`t_l_log_player_level`	等级变化	是	-
`t_l_log_itemlog`	道具日志	是	-
`t_l_log_item`	道具日志	-	是
`t_l_log_money`	货币日志	-	是

14. 完整场景示例

场景 1：运营日报（DAU + 收入 + 付费率 + 留存）

import requests
from datetime import date, timedelta

BASE = "http://dc.example.com:17030/api/v1/external"
HEADERS = {"X-API-Key": "<DC_API_KEY>", "Content-Type": "application/json"}


def daily_report(game_id: str, days: int = 7):
    """生成运营日报"""
    end = date.today()
    start = end - timedelta(days=days - 1)

    # 一站式获取核心指标
    resp = requests.post(
        f"{BASE}/aggregate",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "metrics": [
                "dau", "new_users", "revenue", "pay_users",
                "pay_rate", "arpu", "arppu", "online_peak",
                "retention_d1", "retention_d7",
            ],
            "start_date": start.isoformat(),
            "end_date": end.isoformat(),
        },
    )
    resp.raise_for_status()
    report = {m["metric"]: m["value"] for m in resp.json()["metrics"]}

    print(f"=== {game_id} 运营日报 ({start} ~ {end}) ===")
    print(f"DAU:        {report['dau']:>10,.0f}")
    print(f"新增用户:   {report['new_users']:>10,.0f}")
    print(f"收入:       {report['revenue']:>10,.2f}")
    print(f"付费人数:   {report['pay_users']:>10,.0f}")
    print(f"付费率:     {report['pay_rate']:>10.2%}")
    print(f"ARPU:       {report['arpu']:>10.2f}")
    print(f"ARPPU:      {report['arppu']:>10.2f}")
    print(f"在线峰值:   {report['online_peak']:>10,.0f}")
    print(f"次日留存:   {report['retention_d1']:>10.2%}")
    print(f"7日留存:    {report['retention_d7']:>10.2%}")
    return report


daily_report("game_a", 7)

场景 2：用户活跃度分析

def user_activity_analysis(game_id: str, date_str: str):
    """分析某天用户活跃度"""
    next_day = (date.fromisoformat(date_str) + timedelta(days=1)).isoformat()

    # 1. 分时登录分布（用 free-query 聚合模式）
    print("=== 分时登录分布 ===")
    resp = requests.post(
        f"{BASE}/free-query",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "table": "t_l_log_player_login",
            "filters": {
                "record_date_start": date_str,
                "record_date_end": next_day,
            },
            "group_by": ["HOUR(recorddate)"],
            "aggregates": {
                "login_count": "COUNT(*)",
                "unique_users": "COUNT(DISTINCT openid)",
            },
            "limit": 24,
        },
    )
    resp.raise_for_status()
    for row in resp.json()["data"]:
        hour = row["hour(recorddate)"]
        bar = "#" * (int(float(row["login_count"])) // 1000)
        print(f"  {hour:>2}时: {bar} {row['login_count']}次 / {row['unique_users']}人")

    # 2. 活跃用户等级分布
    print("\n=== 活跃用户等级 Top 10 ===")
    resp = requests.post(
        f"{BASE}/free-query",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "table": "t_l_log_player_login",
            "filters": {
                "record_date_start": date_str,
                "record_date_end": next_day,
            },
            "group_by": ["level"],
            "aggregates": {
                "user_count": "COUNT(DISTINCT openid)",
                "avg_level": "AVG(level)",
            },
            "limit": 10,
        },
    )
    resp.raise_for_status()
    for row in resp.json()["data"]:
        print(f"  Lv.{row['level']}: {row['user_count']}人")

    # 3. 在线时长分析（登出记录）
    print("\n=== 在线时长分布 ===")
    resp = requests.post(
        f"{BASE}/sql-query",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "sql": f"""SELECT
                CASE
                    WHEN online_time < 300 THEN '5分钟以下'
                    WHEN online_time < 1800 THEN '5-30分钟'
                    WHEN online_time < 3600 THEN '30分钟-1小时'
                    WHEN online_time < 7200 THEN '1-2小时'
                    ELSE '2小时以上'
                END AS duration_range,
                COUNT(*) AS user_count
            FROM {_get_db(game_id)}.t_l_log_player_logout
            WHERE recorddate >= '{date_str}' AND recorddate < '{next_day}'
            GROUP BY 1 ORDER BY 1
            LIMIT 10""",
        },
    )
    resp.raise_for_status()
    result = resp.json()
    for row in result["rows"]:
        print(f"  {row[0]}: {row[1]}人")


def _get_db(game_id: str) -> str:
    return {"game_a": "game_a_database", "game_b": "game_b_database"}.get(game_id, "")


user_activity_analysis("game_a", "2026-05-18")

场景 3：付费漏斗分析

def payment_funnel(game_id: str, start: str, end: str):
    """付费漏斗：活跃 -> 付费 -> ARPPU"""
    print(f"=== {game_id} 付费漏斗 ({start} ~ {end}) ===")

    # 1. 总体指标
    resp = requests.post(
        f"{BASE}/aggregate",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "metrics": ["dau", "pay_users", "pay_rate", "revenue", "arpu", "arppu"],
            "start_date": start,
            "end_date": end,
        },
    )
    resp.raise_for_status()
    kpis = {m["metric"]: m["value"] for m in resp.json()["metrics"]}

    print(f"活跃用户: {kpis['dau']:,.0f}")
    print(f"    v 付费率 {kpis['pay_rate']:.2%}")
    print(f"付费用户: {kpis['pay_users']:,.0f}")
    print(f"    v ARPPU ¥{kpis['arppu']:.2f}")
    print(f"总收入: ¥{kpis['revenue']:,.2f}")

    # 2. 每日付费趋势
    resp = requests.post(
        f"{BASE}/free-query",
        headers=HEADERS,
        json={
            "game_id": game_id,
            "table": "t_l_log_charge",
            "filters": {"record_date_start": start, "record_date_end": end},
            "group_by": ["count_date"],
            "aggregates": {
                "daily_revenue": "SUM(money)",
                "pay_users": "COUNT(DISTINCT openid)",
                "order_count": "COUNT(*)",
            },
        },
    )
    resp.raise_for_status()
    print("\n=== 每日付费详情 ===")
    for row in resp.json()["data"]:
        revenue = float(row["daily_revenue"])
        users = int(float(row["pay_users"]))
        orders = int(float(row["order_count"]))
        print(f"  {row['count_date']}: ¥{revenue:>10,.2f}  {users:>5}人  {orders:>5}单  "
              f"人均¥{revenue/users:.2f}" if users else "N/A")


payment_funnel("game_a", "2026-05-12", "2026-05-18")

15. 附录

A. 服务端配置参考

参数	默认值	说明
`RATE_LIMIT_EXTERNAL_QUERY_PER_MINUTE`	10	每分钟每 Key 最大查询次数
`EXTERNAL_QUERY_REDIS_DB`	2	查询结果缓存 Redis db
`EXTERNAL_QUERY_RESULT_TTL`	3600	结果缓存 TTL（秒）
`DW_CONCURRENT_QUERIES`	5	DW 最大并发查询数
`DW_QUERY_TIMEOUT`	60	DW 查询超时（秒）

B. 客户端约束汇总

参数	最小值	最大值	默认值
`limit` (free-query/冷查询)	1	10000	1000
sql-query 自动 LIMIT	-	100000	10000
`cache_ttl`	0	86400	3600
时间范围	1 天	90 天	-
aggregate metrics	1	20	-
sql 长度	1	32768	-

C. 与 Cube.js 的对比

特性	Cube.js	aggregate	free-query	sql-query	冷查询
端点	`:17001/api/v1/load`	`:17030/.../aggregate`	`:17030/.../free-query`	`:17030/.../sql-query`	`:17030/.../query`
认证头	`Authorization`	`X-API-Key`	`X-API-Key`	`X-API-Key`	`X-API-Key`
响应速度	`<500ms`	`<500ms`	1-10s	1-10s	分钟后级
实时数据	-	当日事件表	指标模式可支持	指标/GA 模式可支持	-
SQL 能力	预定义 Cube	无	基础聚合	完整 SQL	预定义数据集
适用场景	固定看板	实时大盘	探索分析	复杂分析	批量报表

目录​

1. 接入信息​

API 端点总览​

2. 认证与权限​

2.1 认证方式​

2.2 Scope 权限模型​

2.3 认证失败响应​

3. 接口一：聚合指标查询​

POST /api/v1/external/aggregate​

请求参数​

支持的全部指标​

数据源自动路由​

响应示例​

cURL 示例​

Python 示例​

4. 接口二：自由查询​

POST /api/v1/external/free-query​

4.1 明细查询​

4.2 聚合查询​

4.3 SDK 中台表自动隔离​

4.4 cURL 示例​

4.5 Python 封装示例​

5. 接口三：标准 SQL 查询​

POST /api/v1/external/sql-query​

请求参数​

7 层安全校验​

响应​

cURL 示例​

Python 封装示例​

6. 接口四：冷查询（异步）​

6.1 POST /api/v1/external/query - 提交冷查询​

6.2 GET /api/v1/external/query/{job_id} - 轮询结果​

6.3 POST /api/v1/external/query/{job_id}/cancel - 取消查询​

Python 冷查询封装​

7. 接口五：异步导出​

7.1 POST /api/v1/external/export - 提交导出任务​

7.2 GET /api/v1/external/export/{task_id} - 查询导出状态​

Python 导出封装​

8. 接口选择决策树​

各接口对比一览​

9. 安全约束​

9.1 SQL 注入防护​

9.2 通用约束​

10. 限流机制​

11. 错误码汇总​

12. SDK 中台表说明​

常用 SDK 表​

13. 各业务数据表​

13.1 业务标识​

13.2 字段风格差异​

13.3 核心表速查​

14. 完整场景示例​

场景 1：运营日报（DAU + 收入 + 付费率 + 留存）​

场景 2：用户活跃度分析​

场景 3：付费漏斗分析​

15. 附录​

A. 服务端配置参考​

B. 客户端约束汇总​

C. 与 Cube.js 的对比​

目录

1. 接入信息

API 端点总览

2. 认证与权限

2.1 认证方式

2.2 Scope 权限模型

2.3 认证失败响应

3. 接口一：聚合指标查询

POST /api/v1/external/aggregate

请求参数

支持的全部指标

数据源自动路由

响应示例

cURL 示例

Python 示例

4. 接口二：自由查询

POST /api/v1/external/free-query

4.1 明细查询

4.2 聚合查询

4.3 SDK 中台表自动隔离

4.4 cURL 示例

4.5 Python 封装示例

5. 接口三：标准 SQL 查询

POST /api/v1/external/sql-query

请求参数

7 层安全校验

响应

cURL 示例

Python 封装示例

6. 接口四：冷查询（异步）

6.1 POST /api/v1/external/query - 提交冷查询

6.2 GET /api/v1/external/query/{job_id} - 轮询结果

6.3 POST /api/v1/external/query/{job_id}/cancel - 取消查询

Python 冷查询封装

7. 接口五：异步导出

7.1 POST /api/v1/external/export - 提交导出任务

7.2 GET /api/v1/external/export/{task_id} - 查询导出状态

Python 导出封装

8. 接口选择决策树

各接口对比一览

9. 安全约束

9.1 SQL 注入防护

9.2 通用约束

10. 限流机制

11. 错误码汇总

12. SDK 中台表说明

常用 SDK 表

13. 各业务数据表

13.1 业务标识

13.2 字段风格差异

13.3 核心表速查

14. 完整场景示例

场景 1：运营日报（DAU + 收入 + 付费率 + 留存）

场景 2：用户活跃度分析

场景 3：付费漏斗分析

15. 附录

A. 服务端配置参考

B. 客户端约束汇总

C. 与 Cube.js 的对比