如何在dbt中实现Snowflake Python UDF并应用至数据模型？

一、在dbt中创建Python UDF #

推荐用sql文件管理自定义函数，适配dbt的版本控制与部署流程：

1. 在dbt项目的models目录下新建udfs子目录，用于存放函数文件
2. 在udfs目录下创建sha3_512.sql，内容如下：

create or replace function "{{ target.schema }}"."sha3_512"(str varchar)
returns varchar
language python
runtime_version = '3.8'
handler = 'hash'
as
$$
import hashlib
 
def hash(str):
    hash_sha3_512 = hashlib.new("sha3_512", str.encode())
    return hash_sha3_512.hexdigest()
$$;

用{{ target.schema }}替代硬编码的schema，可让函数自动部署到dbt当前目标环境的schema，适配开发/生产等多环境场景。若需固定schema，直接替换为指定名称即可。

3. 在dbt_project.yml中配置函数文件的物化方式：

models:
  你的项目名称:
    udfs:
      materialized: ephemeral

二、在模型中调用函数 #

修改你的模型文件，直接调用已创建的函数：

{{ config(materialized = 'view') }}

WITH SEC AS(
    SELECT 
         A."AccountID" AS AccountID,
         A."AccountName" AS AccountName , 
         A."Password" AS Passwords,
         "{{ target.schema }}".sha3_512(A."Password") AS SHash
    FROM {{ ref('Green', 'Account') }} A
   )

SELECT * 
FROM SEC

若函数部署到固定schema（如Unity），直接写"Unity".sha3_512(A."Password")即可。

三、部署与验证 #

1. 先执行dbt run --models udfs.sha3_512创建函数
2. 再执行dbt run --models 你的模型名称运行模型
3. 登录Snowflake，检查目标schema下的函数是否存在，同时验证模型视图中SHash字段的哈希值是否生成正常

注意事项 #

• 确保dbt使用的Snowflake角色拥有CREATE FUNCTION权限，以及Python runtime的相关资源权限
• 首次在Snowflake中使用Python UDF时，需确认仓库已启用Snowpark相关配置
• 运行dbt时，目标schema需与函数部署的schema一致，可通过dbt run --target 环境名切换环境

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