BigQuery含连字符项目中dbt增量Python模型如何使用dbt.this

问题重现 #

在BigQuery环境下实现dbt增量Python模型时，由于项目名称marketing-analytics包含连字符，直接使用dbt.this生成查询语句会触发标识符未加引号的报错：

Possibly unquoted identifier marketing-analytics detected. Please consider quoting with backquotes `marketing-analytics`

核心出错代码：

if dbt.is_incremental:
    # 无法运行
    max_from_this = f"select max(updated_at_new) from {dbt.this}"
    df_raw = dbt.ref("interesting_data").filter(
      F.col("updated_at_new") >= session.sql(max_from_this).collect()[0][0]
    )

尝试手动添加反引号包裹dbt.this或拆分命名空间时，分别出现以下问题：

• 包裹整表名：Table or view not found \marketing-analytics.test_dataset.posts``
• 拆分命名空间包裹：spark_catalog requires a single-part namespace, but got [marketing-analytics, test_dataset]

解决方案 #

方法1：使用dbt内置适配函数生成正确引号的表名 #

dbt提供了适配不同数仓的引号处理逻辑，通过内置方法自动生成符合BigQuery规则的带引号表名：

from dbt.adapters.factory import get_adapter

if dbt.is_incremental:
    adapter = get_adapter(dbt.config)
    # 自动生成适配BigQuery的带引号完整表名
    quoted_table = adapter.quote_as_configured(dbt.this, "identifier")
    max_from_this = f"select max(updated_at_new) from {quoted_table}"
    df_raw = dbt.ref("interesting_data").filter(
        F.col("updated_at_new") >= session.sql(max_from_this).collect()[0][0]
    )

方法2：通过get_relation获取标准格式表对象 #

利用dbt的get_relation方法获取当前模型的关系对象，该对象的字符串输出已适配BigQuery的引号要求：

if dbt.is_incremental:
    this_relation = dbt.get_relation(database=dbt.this.database, schema=dbt.this.schema, identifier=dbt.this.identifier)
    max_from_this = f"select max(updated_at_new) from {this_relation}"
    df_raw = dbt.ref("interesting_data").filter(
        F.col("updated_at_new") >= session.sql(max_from_this).collect()[0][0]
    )

方法3：手动拆分命名空间单独加引号 #

如果需要手动控制格式，确保项目、数据集、表名每个部分单独用反引号包裹：

if dbt.is_incremental:
    # 每个命名空间部分单独添加反引号
    quoted_db = f"`{dbt.this.database}`"
    quoted_schema = f"`{dbt.this.schema}`"
    quoted_table = f"`{dbt.this.identifier}`"
    full_table_name = f"{quoted_db}.{quoted_schema}.{quoted_table}"
    max_from_this = f"select max(updated_at_new) from {full_table_name}"
    df_raw = dbt.ref("interesting_data").filter(
        F.col("updated_at_new") >= session.sql(max_from_this).collect()[0][0]
    )

说明 #

• 方法1、2依赖dbt内置适配逻辑，兼容性更强，优先推荐使用；
• 方法3适合需要自定义表名格式的场景，需注意每个命名空间部分都单独包裹反引号。

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