如何在BigQuery SELECT上下文执行动态SQL计算集中定义的KPI？

方案一：编译阶段用Jinja动态生成KPI列（推荐，无lineage问题） #

因为你的dbt seed是相对静态的，完全可以在dbt模型编译时直接读取seed中的KPI定义，动态生成对应的计算列，不需要运行时动态SQL。

步骤：

1. 确保seed表已通过dbt seed run加载到BigQuery
2. 在目标dbt模型中，用Jinja读取seed的KPI数据：

{% set kpis = dbt_utils.get_column_values(
    ref('kpi_definitions'),
    'kpi_name'
) %}
{% set kpi_expr_map = {} %}
{% for kpi in kpis %}
    {% set expr = dbt_utils.get_column_values(
        ref('kpi_definitions'),
        'kpi_sql_expr',
        where="kpi_name = '" ~ kpi ~ "'"
    )[0] %}
    {% do kpi_expr_map.update({kpi: expr}) %}
{% endfor %}

3. 在SELECT语句中循环生成KPI计算列：

WITH main_data AS (
    SELECT * FROM {{ ref('your_bigquery_model') }}
)
SELECT
    main_data.*,
    {% for kpi_name, expr in kpi_expr_map.items() %}
        {{ expr }} AS {{ kpi_name }}{% if not loop.last %},{% endif %}
    {% endfor %}
FROM main_data
JOIN {{ ref('kpi_definitions') }} kpis
    ON main_data.关联键 = kpis.关联键
GROUP BY main_data.*

优缺点：

• 优点：完全贴合dbt编译逻辑，lineage自动识别主模型与seed的依赖，无动态SQL性能损耗
• 缺点：KPI更新后需重新执行dbt seed run+dbt run，不适合KPI实时变动场景

方案二：运行时生成JSON格式KPI结果（适合KPI频繁变动） #

如果需要支持KPI实时更新（无需重新编译dbt模型），可结合BigQuery存储过程实现，同时手动维护dbt lineage依赖。

步骤：

1. 创建BigQuery存储过程，负责动态计算KPI并生成JSON结果：

CREATE OR REPLACE PROCEDURE `project.dataset.generate_kpi_json`(
    main_table STRING,
    kpi_table STRING,
    output_table STRING
)
BEGIN
    DECLARE kpi_json_expr STRING;
    -- 拼接所有KPI的JSON键值对表达式
    SELECT STRING_AGG(
        FORMAT("'%s': %s", kpi_name, kpi_sql_expr),
        ', '
    ) INTO kpi_json_expr
    FROM `project.dataset.kpi_definitions`;

    -- 执行动态SQL生成结果表
    EXECUTE IMMEDIATE FORMAT("""
        CREATE OR REPLACE TABLE `%s` AS
        SELECT
            m.*,
            JSON_OBJECT(%s) AS kpi_results
        FROM `%s` m
        JOIN `%s` k
            ON m.关联键 = k.关联键
        GROUP BY m.*
    """, output_table, kpi_json_expr, main_table, kpi_table);
END;

2. 在dbt模型的post-hook中调用存储过程，同时手动声明seed依赖：

{{ config(
    post_hook=[
        "CALL `project.dataset.generate_kpi_json`('{{ ref('your_bigquery_model') }}', '{{ ref('kpi_definitions') }}', '{{ this }}')"
    ],
    meta={
        "dbt_dependencies": ["kpi_definitions"]
    }
) }}

-- 生成空基表保证dbt识别主模型依赖
SELECT * FROM {{ ref('your_bigquery_model') }} LIMIT 0

优缺点：

• 优点：KPI更新后仅需重新执行dbt run，无需编译模型，适配频繁变动场景
• 缺点：dbt默认lineage无法自动识别存储过程依赖，需手动在meta中声明，性能略低于静态列方案

方案三：动态生成透视表（适合列格式输出且KPI变动频繁） #

结合编译时Jinja读取与运行时动态透视，兼顾动态性与列格式输出需求。

1. 编写dbt宏封装透视逻辑：

{% macro generate_kpi_pivot(main_model, kpi_seed, join_key) %}
    {% set kpis = dbt_utils.get_column_values(ref(kpi_seed), 'kpi_name') %}
    {% set pivot_cols = [] %}
    {% for kpi in kpis %}
        {% do pivot_cols.append(FORMAT("MAX(CASE WHEN kpi_name = '%s' THEN %s END) AS %s", kpi, kpi, kpi)) %}
    {% endfor %}

    EXECUTE IMMEDIATE FORMAT("""
        SELECT
            m.*,
            %s
        FROM {{ ref(main_model) }} m
        JOIN {{ ref(kpi_seed) }} k
            ON m.{{ join_key }} = k.{{ join_key }}
        GROUP BY m.*
    """, '{{ pivot_cols | join(", ") }}');
{% endmacro %}

2. 在dbt模型中调用宏生成结果表：

CREATE OR REPLACE TABLE {{ this }} AS
{{ generate_kpi_pivot('your_bigquery_model', 'kpi_definitions', '关联键') }}

优缺点：

• 优点：兼顾动态透视与列格式输出，dbt可自动识别主模型与seed的依赖
• 缺点：KPI更新后需重新编译模型，但比方案一更灵活

内容的提问来源于stack exchange，提问作者Adam Cunnington