能否让DBT以批量方式运行数据转换任务？

针对大表转换时资源消耗大、易失败的问题，完全可以通过DBT自身功能实现批量处理，无需手动导出SQL循环执行。结合你提到的「单源表、无聚合/窗口函数」的场景，推荐以下两种高效方案：

方案1：基于主键范围的增量模型（推荐） #

利用DBT的增量模型特性，每次加载一个批次的数据，通过主键（如id）范围避免低效的OFFSET。这种方式性能更优，且天然支持断点续传。

步骤1：编写增量模型文件 #

创建models/large_table_transform.sql：

{{ config(
    materialized='incremental',
    unique_key='id'  -- 替换为你的源表主键
) }}

-- 这里可以添加你的转换逻辑（无聚合场景下直接SELECT或简单字段转换即可）
SELECT
    id,
    col1,
    col2,
    -- 其他字段或转换逻辑
FROM {{ source('your_source_schema', 'large_source_table') }}
{% if is_incremental() %}
-- 增量加载：只取比目标表中最大id更大的数据
WHERE id > (SELECT COALESCE(MAX(id), 0) FROM {{ this }})
{% endif %}
-- 每个批次加载的行数
LIMIT {{ var('batch_size', 10000) }}

步骤2：编写自动批量执行的宏 #

创建macros/run_all_batches.sql，让DBT自动计算批次数量并循环执行：

{% macro run_all_batches(model_name, batch_size=10000) %}
    -- 计算源表总记录数
    {% set count_query %}
        SELECT COUNT(*) FROM {{ source('your_source_schema', 'large_source_table') }}
    {% endset %}
    {% set total_rows = run_query(count_query).columns[0][0] %}
    {% set total_batches = (total_rows / batch_size)|round(method='ceil')|int %}

    {% do log(f"待处理总记录数: {total_rows}, 总批次: {total_batches}", info=True) %}

    {% for batch_num in range(total_batches) %}
        {% do log(f"正在处理第 {batch_num + 1}/{total_batches} 批次", info=True) %}
        {% set run_command = "dbt run --select " ~ model_name ~ " --vars '{\"batch_size\": " ~ batch_size ~ "}'" %}
        {% do run_command(run_command) %}
    {% endfor %}
{% endmacro %}

步骤3：执行批量转换 #

运行以下命令，DBT会自动完成所有批次的处理：

dbt run-operation run_all_batches --args '{"model_name": "large_table_transform", "batch_size": 10000}'

方案2：基于OFFSET的批量模型（无主键时用） #

如果源表没有主键，只能用LIMIT + OFFSET实现分页。同样通过宏自动循环执行：

步骤1：编写参数化模型文件 #

创建models/large_table_offset_transform.sql：

{{ config(
    materialized='append'  -- 每次运行追加数据
) }}

SELECT
    col1,
    col2,
    -- 其他字段或转换逻辑
FROM {{ source('your_source_schema', 'large_source_table') }}
LIMIT {{ var('batch_size', 10000) }}
OFFSET {{ var('batch_size', 10000) * var('batch_num', 0) }}

步骤2：编写自动批量执行的宏 #

创建macros/run_offset_batches.sql：

{% macro run_offset_batches(model_name, batch_size=10000) %}
    {% set count_query %}
        SELECT COUNT(*) FROM {{ source('your_source_schema', 'large_source_table') }}
    {% endset %}
    {% set total_rows = run_query(count_query).columns[0][0] %}
    {% set total_batches = (total_rows / batch_size)|round(method='ceil')|int %}

    {% do log(f"待处理总记录数: {total_rows}, 总批次: {total_batches}", info=True) %}

    {% for batch_num in range(total_batches) %}
        {% do log(f"正在处理第 {batch_num + 1}/{total_batches} 批次", info=True) %}
        {% set run_command = "dbt run --select " ~ model_name ~ " --vars '{\"batch_size\": " ~ batch_size ~ ", \"batch_num\": " ~ batch_num ~ "}'" %}
        {% do run_command(run_command) %}
    {% endfor %}
{% endmacro %}

步骤3：执行批量转换 #

运行命令：

dbt run-operation run_offset_batches --args '{"model_name": "large_table_offset_transform", "batch_size": 10000}'

注意事项 #

• 优先使用方案1，OFFSET在大表上会扫描前面所有行，性能远低于主键范围过滤。
• 若需要全量重跑，先执行dbt run --select <model_name> --full-refresh清空目标表，再运行批量宏。
• 可以根据数据库资源调整batch_size，避免单个批次消耗过多资源。

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