dbt-snowflake环境下获取模型行数的替代方案问询

针对你的dbt-snowflake 1.5.2 + dbt_artifacts 2.5.0环境，以下几种方案可以替代旧版dbt_artifacts的row_count收集功能：

方案一：基于Snowflake系统视图的自定义模型 #

直接利用Snowflake的INFORMATION_SCHEMA元数据视图快速获取表的行数（视图需单独处理），无需额外包依赖。

创建一个元数据模型（比如models/meta/table_row_counts.sql）：

{{ config(
    materialized='table',
    schema='meta'
) }}

-- 收集BASE TABLE的行数（Snowflake缓存数据，默认24小时刷新）
SELECT
    table_catalog AS database_name,
    table_schema,
    table_name,
    row_count,
    last_altered AS statistics_updated_at
FROM {{ target.database }}.INFORMATION_SCHEMA.TABLES
WHERE table_type = 'BASE TABLE'

UNION ALL

-- 收集VIEW的行数（注意：大视图执行COUNT(*)会消耗较多资源，按需启用）
SELECT
    table_catalog AS database_name,
    table_schema,
    table_name,
    (SELECT COUNT(*) FROM {{ target.database }}.{{ table_schema }}.{{ table_name }}) AS row_count,
    CURRENT_TIMESTAMP() AS statistics_updated_at
FROM {{ target.database }}.INFORMATION_SCHEMA.TABLES
WHERE table_type = 'VIEW'

注意：INFORMATION_SCHEMA.TABLES中的row_count是缓存值，若需要实时表行数，可将表部分的row_count替换为(SELECT COUNT(*) FROM ...)，但会增加查询成本。

方案二：使用dbt-snowflake-utils包的宏 #

借助dbt-snowflake-utils提供的get_relation_row_count宏，灵活获取单个模型的行数，再通过dbt_utils合并结果。

1. 先在packages.yml中添加依赖：

packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.0.0", "<2.0.0"]
  - package: snowflake-labs/dbt_snowflake_utils
    version: [">=0.1.0", "<1.0.0"]

执行dbt deps安装依赖。

2. 创建宏macros/get_model_row_count.sql：

{% macro get_model_row_count(relation) %}
    SELECT
        '{{ relation.database }}' AS database_name,
        '{{ relation.schema }}' AS schema_name,
        '{{ relation.name }}' AS model_name,
        {{ dbt_snowflake_utils.get_relation_row_count(relation) }} AS row_count,
        CURRENT_TIMESTAMP() AS loaded_at
    FROM DUAL
{% endmacro %}

3. 创建合并所有模型行数的模型：

{{ config(
    materialized='table',
    schema='meta'
) }}

{% set models = graph.nodes.values() | selectattr('resource_type', 'equalto', 'model') | list %}
{% set queries = [] %}

{% for model in models %}
    {% do queries.append(get_model_row_count(model)) %}
{% endfor %}

{{ dbt_utils.union_sql(queries) }}

方案三：通过dbt运行钩子自动收集行数 #

利用dbt的on-run-end钩子，在每个模型运行完成后自动收集行数并插入自定义元数据表，保证数据实时性。

1. 创建元数据表模型models/meta/model_row_counts.sql：

{{ config(
    materialized='table',
    schema='meta'
) }}

CREATE OR REPLACE TABLE {{ this }} (
    database_name VARCHAR,
    schema_name VARCHAR,
    model_name VARCHAR,
    row_count NUMBER,
    run_timestamp TIMESTAMP,
    status VARCHAR
)

2. 创建钩子宏macros/insert_row_count_on_run_end.sql：

{% macro insert_row_count_on_run_end() %}
    {% for result in results %}
        {% if result.resource_type == 'model' %}
            {% set relation = adapter.get_relation(
                database=result.database,
                schema=result.schema,
                identifier=result.name
            ) %}
            {% if relation %}
                {% set row_count = dbt_snowflake_utils.get_relation_row_count(relation) %}
                {% set insert_query %}
                    INSERT INTO {{ target.database }}.meta.model_row_counts (
                        database_name, schema_name, model_name, row_count, run_timestamp, status
                    ) VALUES (
                        '{{ result.database }}', '{{ result.schema }}', '{{ result.name }}',
                        {{ row_count }}, CURRENT_TIMESTAMP(), '{{ result.status }}'
                    )
                {% endset %}
                {% do run_query(insert_query) %}
            {% endif %}
        {% endif %}
    {% endfor %}
{% endmacro %}

3. 在dbt_project.yml中配置钩子：

on-run-end:
  - "{{ insert_row_count_on_run_end() }}"

注意：大模型执行COUNT(*)会延长dbt运行时间，可根据模型优先级选择性收集。

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