在DBT(BigQuery)中为动态列模型自动生成YML文件

问题背景 #

• 已基于存储嵌套JSON的源表events，通过dbt_utils.pivot函数将动态变化的name字段值转为order_details模型的列，SQL脚本运行正常
• 需求：不借助Python等外部语言，仅通过DBT自动生成符合指定结构的YML文件，列名禁止硬编码

目标YML结构 #

models:
  - name: orders
    description: '{{ doc("orders") }}'
    tests:
      - unique:
          column_name: "order_number"
    columns:
      - name: code
        description: '{{ doc("code") }}'
      - name: order_date
        description: '{{ doc("order_date") }}'

当前order_details模型SQL脚本 #

select
    order_number,
    order_status_time,
    order_status,
    {{
        dbt_utils.pivot(
            "name",
            dbt_utils.get_column_values(
                    ref("events"),
                    "name"
            ),
            then_value="creation_time",
            else_value="NULL",
            agg="max"
        )
    }}
from  {{ ref("events") }}
{{ dbt_utils.group_by(3) }}

实现方案 #

1. 编写自定义宏获取模型列 #

在项目的macros/目录下创建get_model_columns.sql宏，用于动态获取目标模型的所有列名：

{% macro get_model_columns(model_name) %}
    {% set relation = ref(model_name) %}
    {% set columns = adapter.get_columns_in_relation(relation) %}
    {% set column_names = columns | map(attribute='name') | list %}
    {{ return(column_names) }}
{% endmacro %}

2. 生成动态YML文件 #

在models/目录下创建generated_order_details.yml，通过Jinja2循环自动渲染列配置：

models:
  - name: order_details
    description: '{{ doc("order_details") }}'
    tests:
      - unique:
          column_name: "order_number"
    columns:
      {% for column in get_model_columns('order_details') %}
      - name: {{ column }}
        description: '{{ doc(column) if doc_exists(column) else "动态生成的订单详情列" }}'
      {% endfor %}

关键说明 #

• 利用DBT内置的adapter.get_columns_in_relation方法直接读取order_details模型的列信息，完全适配pivot生成的动态列
• 通过Jinja循环遍历列名，自动生成每个列的YML节点
• 文档引用{{ doc(column) }}会关联你在docs/目录下定义的对应列文档；如果未定义文档，用doc_exists判断并返回默认描述

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