美团如何用指标平台统一数据口径、加速查询——新一代 BI 架构实践

# metrics/order_amount.yaml — 指标平台中的指标定义
metric:
  name: order_amount
  display_name: 订单金额
  business_caliber: 所有已完成订单的金额总和，不含退款
  measure:
    expr: SUM(order_amount)
    filter: order_status = 'completed'
  dimensions:
    - city_id
    - business_line
    - order_date
  source_table: dw.order_detail
  pre_aggregate:
    enabled: true
    granularity: [city_id, business_line, order_date]
    schedule: daily at 02:00

# semantic_query_generator.py
import yaml
from pathlib import Path

def load_metric(metric_name: str) -> dict:
    """从指标平台加载指标定义"""
    path = Path(f"metrics/{metric_name}.yaml")
    with open(path) as f:
        return yaml.safe_load(f)["metric"]

def generate_query(metric_name: str, dimensions: list[str],
                   filters: dict | None = None) -> str:
    """根据指标语义定义自动生成 SQL"""
    metric = load_metric(metric_name)
    measure_expr = metric["measure"]["expr"]
    measure_filter = metric["measure"]["filter"]
    source = metric["source_table"]

    # 维度列拼接
    dim_cols = ", ".join(dimensions)
    # 合并指标内置过滤与用户额外过滤
    where_parts = [measure_filter]
    if filters:
        for k, v in filters.items():
            where_parts.append(f"{k} = '{v}'")
    where_clause = " AND ".join(where_parts)

    sql = f"""
SELECT
    {dim_cols},
    {measure_expr} AS {metric_name}
FROM {source}
WHERE {where_clause}
GROUP BY {dim_cols}
"""
    return sql.strip()

# ---- 使用示例 ----
if __name__ == "__main__":
    # 业务方只需要指定：查什么指标、按什么维度、加什么过滤
    # 口径和计算逻辑由指标平台自动注入
    query = generate_query(
        metric_name="order_amount",
        dimensions=["city_id", "order_date"],
        filters={"business_line": "外卖"}
    )
    print(query)
    # 输出:
    # SELECT city_id, order_date,
    #        SUM(order_amount) AS order_amount
    # FROM dw.order_detail
    # WHERE order_status = 'completed' AND business_line = '外卖'
    # GROUP BY city_id, order_date

pip install pyyaml

# pre_aggregate_router.py
def should_use_pre_aggregate(metric: dict, query_dims: list[str]) -> bool:
    """判断查询是否可以走预聚合表"""
    pre = metric.get("pre_aggregate", {})
    if not pre.get("enabled"):
        return False
    # 查询维度必须是预聚合粒度的子集
    pre_dims = set(pre["granularity"])
    query_dim_set = set(query_dims)
    return query_dim_set.issubset(pre_dims)

def generate_pre_aggregate_query(metric_name: str,
                                  dimensions: list[str],
                                  filters: dict | None = None) -> str:
    """生成走预聚合表的 SQL"""
    metric = load_metric(metric_name)
    pre_table = f"pre_agg.{metric_name}"  # 预聚合表命名规则
    dim_cols = ", ".join(dimensions)
    where_parts = []
    if filters:
        for k, v in filters.items():
            # 只保留查询维度中存在的过滤
            if k in metric["pre_aggregate"]["granularity"]:
                where_parts.append(f"{k} = '{v}'")
    where_clause = " AND ".join(where_parts) if where_parts else "1=1"

    sql = f"""
SELECT
    {dim_cols},
    {metric_name}
FROM {pre_table}
WHERE {where_clause}
GROUP BY {dim_cols}
"""
    return sql.strip()

# ---- 使用示例：智能路由 ----
if __name__ == "__main__":
    metric = load_metric("order_amount")
    dims = ["city_id", "order_date"]

    if should_use_pre_aggregate(metric, dims):
        print("走预聚合表:")
        print(generate_pre_aggregate_query("order_amount", dims,
                                           {"business_line": "外卖"}))
    else:
        print("走明细表:")
        print(generate_query("order_amount", dims,
                             {"business_line": "外卖"}))