如何在dbt-spark自定义物化中动态转换列解决UNION ALL schema不匹配问题？

问题背景 #

使用dbt-spark适配器开发写入S3 Delta表的自定义混合SCD Type1/Type2物化时，通过UNION ALL对比源临时视图与目标Delta表生成变更标记（Insert/I、Type1 Update/U1等），遇到Spark严格类型检查报错：

• 源表列fnnc_dpnd_flg被推断为STRING类型
• 目标表同名列定义为BOOLEAN类型
• 抛出AnalysisException: [INCOMPATIBLE_COLUMN_TYPE] UNION can only be performed on tables with compatible column types

受环境限制无法修改上游DDL，必须在dbt物化逻辑内解决类型不兼容问题。

解决方案：动态生成带类型转换的源列 #

可以通过dbt的Jinja上下文访问target_relation的schema信息，遍历目标列并对需要转换的源列自动生成CAST语句，替换原有的简单列名列表。

1. 修改source_cols生成逻辑 #

替换原有的{{ source_cols }}定义，改为动态生成带类型转换的列列表：

{% set source_cols = [] %}
{% for col in target_relation.columns %}
    {% set src_col = col.name %}
    {% if col.data_type.lower() == 'boolean' %}
        -- 对目标为BOOLEAN的源列做类型转换，兼容常见字符串值
        {% do source_cols.append("CASE WHEN s." ~ src_col ~ " IN ('Y', 'TRUE', '1', 'YES') THEN TRUE WHEN s." ~ src_col ~ " IN ('N', 'FALSE', '0', 'NO') THEN FALSE ELSE NULL END AS " ~ src_col) %}
    {% else %}
        {% do source_cols.append("s." ~ src_col) %}
    {% endif %}
{% endfor %}
{% set source_cols = source_cols | join(', ') %}

2. 完整修改后的Jinja模板 #

将上述逻辑嵌入原有的自定义物化模板：

{% call statement('create_change_table') %}
    {% set source_cols = [] %}
    {% for col in target_relation.columns %}
        {% set src_col = col.name %}
        {% if col.data_type.lower() == 'boolean' %}
            {% do source_cols.append("CASE WHEN s." ~ src_col ~ " IN ('Y', 'TRUE', '1', 'YES') THEN TRUE WHEN s." ~ src_col ~ " IN ('N', 'FALSE', '0', 'NO') THEN FALSE ELSE NULL END AS " ~ src_col) %}
        {% else %}
            {% do source_cols.append("s." ~ src_col) %}
        {% endif %}
    {% endfor %}
    {% set source_cols = source_cols | join(', ') %}

    CREATE OR REPLACE TABLE {{ tmp_flags }}
    USING {{ table_format }}
    AS
    -- 1. New Inserts
    SELECT
        {{ source_cols }},
        'I' AS flg
    FROM {{ temp_relation }} s
    LEFT JOIN {{ target_relation }} t
        ON {{ join_bk_expr }}
        AND t.ds_record_end_dt = to_date('2400-01-01')
    WHERE
        {% for col in unique_key %}
            t.{{ col }} IS NULL
            {% if not loop.last %} AND {% endif %}
        {% endfor %}

    UNION ALL

    -- 2. Type 1 Updates
    SELECT
        {{ source_cols }},
        'U1' AS flg
    FROM {{ temp_relation }} s
    INNER JOIN {{ target_relation }} t
        ON {{ join_bk_expr }}
        AND t.ds_record_end_dt = to_date('2400-01-01')
    WHERE
        -- (Logic to detect Type 1 changes)
        {% if type1_cols | length > 0 %}
            coalesce({{ type1_src }}, '$') != coalesce({{ type1_tgt }}, '$')
            AND coalesce({{ type2_src }}, '$') = coalesce({{ type2_tgt }}, '$')
        {% else %}
            false
        {% endif %}
    
    -- (Additional UNIONs for U2 and D omitted for brevity)
    ;
{% endcall %}

关键说明 #

• 利用target_relation.columns获取目标表的完整schema信息，确保列名和类型与目标严格对齐
• 针对BOOLEAN类型的转换使用CASE WHEN而非直接CAST，是为了兼容上游常见的字符串BOOLEAN表示（如Y/N、1/0），如果上游是标准的'true'/'false'字符串，直接用CAST(s.{{ src_col }} AS BOOLEAN) AS {{ src_col }}即可
• 可以扩展逻辑处理其他类型不兼容场景（如STRING转DATE、INT），只需增加对应的elif分支判断目标列类型

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