如何在Snowpark中引用dbt定义的数据源？

问题 #

我已经定义了如下dbt数据源：

version: 2

sources:
  - name: mysource
    database: MY_DATABASE
    schema: myschema
    tables:
      - name: table1

在dbt SQL模型中我是这样引用它的：

{%- set source_table = source('mysource', 'table1') -%}

SELECT * 
FROM {{ source_table }} source

（实际场景更复杂，此处仅作演示）

现在我想把这个模型改写成Snowpark Python模型，Snowpark模型的基础框架如下：

def model(dbt, session):
    # Must be either table or incremental (view is not currently supported)
    dbt.config(materialized = "table")

    # DataFrame representing an upstream model
    df = dbt.ref("my_first_dbt_model")
 
    return df

示例中用dbt.ref()引用上游模型，请问在Snowpark模型里该如何引用我之前定义的dbt数据源？

解答 #

在Snowpark Python模型中，你可以使用dbt.source()方法来引用已定义的dbt数据源，用法和SQL模型里的source()宏完全对应，参数同样是数据源名称和表名。

改写后的完整Snowpark模型代码如下：

def model(dbt, session):
    # 设置物化方式为table（目前仅支持table或incremental）
    dbt.config(materialized = "table")

    # 通过dbt.source()引用定义的数据源表
    df = dbt.source("mysource", "table1")
 
    return df

如果需要对数据源表做更复杂的处理（比如添加过滤、转换逻辑），可以基于返回的DataFrame进行操作，例如：

def model(dbt, session):
    dbt.config(materialized = "table")

    df = dbt.source("mysource", "table1")
    # 示例：筛选特定字段并添加过滤条件
    processed_df = df.select("col1", "col2").filter(df["col3"] > 100)
 
    return processed_df

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