将Snowflake游标循环语句转换为dbt模型的技术问询

核心逻辑梳理 #

原Snowflake脚本的作用是：遍历dev.abc表中的所有列名，对dev.facts中rev>0的数据，按每个列名对应的字段分组，统计去重账号数、求和字段值，最终将结果写入dev.results。

原脚本计数不匹配的原因 #

你原脚本里的GROUP BY 1,2,3是语法错误：SELECT列表的第3列是聚合函数COUNT(DISTINCT account)，不能作为GROUP BY的分组依据，这会导致分组逻辑混乱，直接引发计数结果错误。正确的分组应该是GROUP BY 1,2（仅按coh_type和coh分组）。

dbt实现方案（推荐） #

dbt是声明式数据建模工具，不建议用游标这类过程式逻辑，推荐用宏生成动态UNION ALL语句的方式实现，既符合dbt最佳实践，又能避免重复插入等问题。

步骤1：编写dbt宏

在macros/generate_cohorts.sql中创建宏：

{% macro generate_cohorts() %}
    -- 从dev.abc获取所有需要分组的列名
    {% set col_names_query %}
        SELECT col_name FROM {{ ref('abc') }}
    {% endset %}
    {% set col_names = run_query(col_names_query).columns[0].values() %}
    
    -- 遍历每个列名，生成对应的统计查询
    {% for col in col_names %}
        SELECT
            '{{ col }}' AS coh_type,
            {{ col }} AS coh,
            COUNT(DISTINCT account) AS cust,
            SUM(so_ls) AS so_ls,
            SUM(si_ls) AS si_ls
        FROM (
            SELECT
                *,
                LENGTH(num_empl) AS emp_d,
                LENGTH(ann_rev) AS ann_rev_d
            FROM {{ ref('facts') }}
            WHERE rev > 0
        ) AS A
        GROUP BY 1, 2
        -- 除了最后一个查询，其余后面加UNION ALL
        {% if not loop.last %}
            UNION ALL
        {% endif %}
    {% endfor %}
{% endmacro %}

步骤2：创建dbt模型

在models/cohorts_results.sql中创建结果模型：

{{ config(
    materialized='table',
    schema='dev',
    alias='results'
) }}

{{ generate_cohorts() }}

方案优势 #

1. 原子性执行：dbt每次运行都会重新生成整个表（或增量更新），不会像原游标脚本那样重复插入数据，确保结果准确
2. 依赖自动管理：通过ref()函数自动识别abc和facts模型的依赖，执行时会按顺序运行
3. 易于维护：动态生成的SQL比游标逻辑更直观，便于调试和修改
4. 可测试性：可以通过dbt test添加断言，验证统计结果的正确性

注意事项 #

• 确保dev.abc中的col_name都是dev.facts中存在的有效列，否则会触发SQL语法错误
• 如果列数量极大，UNION ALL可能有性能瓶颈，可以考虑结合Snowflake的UNPIVOT功能优化，但对于大多数场景，上述方案足够高效
• 如果需要增量更新（仅处理新增数据），可以将materialized改为incremental，并添加增量过滤逻辑

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