dbt Jinja 巨集動態 SQL 與套件管理

– 9 分鐘閱讀

dbt 提供 Jinja 範本引擎，賦予 SQL 更強大的動態能力。透過 Jinja，我們可以根據變數、迴圈和條件陳述式生成不同的 SQL 查詢，避免重複程式碼並提升靈活性。文章將逐步示範如何使用 Jinja 建立動態 SQL，例如根據支付型別列表生成對應的金額統計查詢。此外，dbt 的巨集功能可以將常用的 SQL 片段封裝成可重複使用的模組，進一步提升程式碼的可維護性。我們將探討如何建立巨集、傳遞引數以及在不同模型中重複使用巨集。文章也將示範如何動態地從資料函式庫中提取支付型別，避免硬編碼，並介紹如何根據執行環境調整資料集範圍，提高開發效率。最後，我們將介紹 dbt 的套件管理機制，如何安裝和使用 dbt_utils 等套件，並簡述 dbt 語義層如何簡化資料模型的管理和演進。

使用 Jinja 進行動態 SQL 查詢

在 dbt 專案中，Jinja 是一種強大的範本引擎，可以用來建立動態 SQL 查詢。以下是一個複雜的使用範例：

{% set payment_types = ['cash', 'credit'] %}
with pivot_and_aggregate_payments_to_order_grain as (
  select 
    order_id,
    {% for payment_type in payment_types %}
    sum(case when payment_type = '{{ payment_type }}' and status = 'success' then amount end) as {{ payment_type }}_amount,
    {% endfor %}
    sum(case when status = 'success' then amount end) as total_amount
  from payments
  group by 1
)
select * from pivot_and_aggregate_payments_to_order_grain

內容解密：

  1. {% set payment_types = ['cash', 'credit'] %}：宣告一個名為 payment_types 的變數，並將其設定為一個包含兩個元素的列表：'cash' 和 'credit'。
  2. {% for payment_type in payment_types %}：使用 for 迴圈迭代 payment_types 列表中的每個元素。
  3. sum(case when payment_type = '{{ payment_type }}' and status = 'success' then amount end) as {{ payment_type }}_amount：計算每個付款型別的總金額，並將結果命名為對應的付款型別加上 _amount 字尾。
  4. {{ ... }}：用於輸出表示式的值，例如 {{ payment_type }} 用於生成對應的付款型別名稱。
  5. {# ... #}：用於新增註解，不會被輸出到最終的 SQL 查詢中。

Jinja 語法要點

  • {% ... %}：用於陳述式，如變數宣告和迴圈控制。
  • {{ ... }}：用於輸出表示式的值。
  • {# ... #}：用於新增註解。
  • {%- ... -%}：使用連字元號控制空白字元的輸出。

使用 SQL 巨集

在 dbt 專案中，巨集是用來自動化重複性任務的程式碼片段。以下是一個簡單的巨集範例，用於計算兩個數字的總和：

{% macro sum(x, y) -%}
select {{ x + y }}
{%- endmacro %}

內容解密：

  1. {% macro sum(x, y) -%}：定義一個名為 sum 的巨集，接受兩個引數 x 和 y。
  2. select {{ x + y }}：計算 x 和 y 的總和，並輸出結果。
  3. {%- endmacro %}：結束巨集定義。

使用巨集

要在 dbt 專案中使用巨集，可以在模型檔案中呼叫它，如下所示：

{{ sum(13, 89) }}

或者，使用變數作為引數：

{% set x = 13 %}
{% set y = 89 %}
{{ sum(x, y) }}

內容解密：

  1. {{ sum(13, 89) }}：呼叫 sum 巨集，傳入 13 和 89 作為引數。
  2. {% set x = 13 %}`` 和 ``{% set y = 89 %}：宣告變數 x 和 y，並將其設定為 13 和 89。
  3. {{ sum(x, y) }}：呼叫 sum 巨集，傳入變數 x 和 y 作為引數。

建立可重複使用的巨集

以下是一個更實際的範例，用於集中管理付款型別組態：

{% macro get_payment_types() %}
{{ return(["cash", "credit"]) }}
{% endmacro %}

內容解密：

  1. {% macro get_payment_types() %}：定義一個名為 get_payment_types 的巨集，不接受任何引數。
  2. {{ return(["cash", "credit"]) }}：傳回一個包含 'cash' 和 'credit' 的列表。

然後，在模型檔案中呼叫這個巨集：

{%- set payment_types = get_payment_types() -%}

圖表翻譯：

此圖示呈現了使用 Jinja 和巨集來建立動態 SQL 查詢的流程。首先，我們定義了一個包含付款型別的列表，然後使用迴圈迭代這個列表，生成對應的 SQL 程式碼。接著，我們定義了一個簡單的巨集，用於計算兩個數字的總和，並展示瞭如何在模型檔案中呼叫這個巨集。最後，我們建立了一個可重複使用的巨集，用於集中管理付款型別組態，並在模型檔案中呼叫它。

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title dbt Jinja 巨集動態 SQL 與套件管理

package "資料庫架構" {
    package "應用層" {
        component [連線池] as pool
        component [ORM 框架] as orm
    }

    package "資料庫引擎" {
        component [查詢解析器] as parser
        component [優化器] as optimizer
        component [執行引擎] as executor
    }

    package "儲存層" {
        database [主資料庫] as master
        database [讀取副本] as replica
        database [快取層] as cache
    }
}

pool --> orm : 管理連線
orm --> parser : SQL 查詢
parser --> optimizer : 解析樹
optimizer --> executor : 執行計畫
executor --> master : 寫入操作
executor --> replica : 讀取操作
cache --> executor : 快取命中

master --> replica : 資料同步

note right of cache
  Redis/Memcached
  減少資料庫負載
end note

@enduml

圖表翻譯： 此圖示展示了從定義付款型別列表到呼叫可重複使用的巨集的整個流程。透過使用 Jinja 和巨集，我們可以建立動態且可重複使用的 SQL 查詢。

提升SQL巨集的動態能力與模組化設計

在前面的範例中，我們觀察到巨集（macro）的功能並不是非常強大或靜態。為了避免依賴手動輸入，我們可以採取以下步驟來最佳化它：

  • 瞭解來源資料，以便動態提取支付型別。
  • 以模組化的思維重新設計巨集。

動態提取支付型別

首先，我們可以使用範例5-13中的查詢陳述式來克服第一點。

範例5-13：取得不同的支付型別查詢

select
    distinct payment_type
from {{ ref('stg_stripe_order_payments') }}
order by 1

執行此查詢後，我們可以看到它包含了不同的支付型別。為了將其轉換為巨集，我們可以將範例5-14中的程式碼複製到get_payment_types.sql檔案中。

範例5-14：使get_payment_types更具動態和擴充套件性

{% macro get_payment_types() %}
    {% set payment_type_query %}
        select
            distinct payment_type
        from {{ ref('stg_stripe_order_payments') }}
        order by 1
    {% endset %}
    {% set results = run_query(payment_type_query) %}
    {% if execute %}
        {# Return the first column #}
        {% set results_list = results.columns[0].values() %}
    {% else %}
        {% set results_list = [] %}
    {% endif %}
    {{ return(results_list) }}
{% endmacro %}

內容解密：

  1. 首先，我們宣告了一個名為payment_type_query的查詢變數。
  2. 接著，使用run_query函式執行查詢，並將結果儲存在results變數中。
  3. 然後，檢查Jinja是否處於執行模式（即SQL正在被執行）。如果是，則將資料集第一列的結果儲存在results_list中。這個第一列包含了不同的值。
  4. 最後，傳回results_list變數，以便在模型中使用。

提升模組化設計

現在，如果我們再次編譯int_payment_type_amount_per_order.sql模型，雖然表面上沒有變化，但我們已經實作了更具擴充套件性的程式碼。目前，當出現新的支付型別時，不需要手動輸入。但是，我們還可以進一步提高模組化程度。

範例5-15：重複使用程式碼以處理不同的場景

{# Generic macro to give a column name and table, outputs the distinct fields of the given column name #}
{% macro get_column_values(column_name, table_name) %}
    {% set relation_query %}
        select distinct
            {{ column_name }}
        from {{ table_name }}
        order by 1
    {% endset %}
    {% set results = run_query(relation_query) %}
    {% if execute %}
        {# Return the first column #}
        {% set results_list = results.columns[0].values() %}
    {% else %}
        {% set results_list = [] %}
    {% endif %}
    {{ return(results_list) }}
{% endmacro %}

{# Macro to get the distinct payment_types #}
{% macro get_payment_types() %}
    {{ return(get_column_values('payment_type', ref('stg_stripe_order_payments'))) }}
{% endmacro %}

{# Macro to get the distinct payment_methods #}
{% macro get_payment_methods() %}
    {{ return(get_column_values('payment_method', ref('stg_stripe_order_payments'))) }}
{% endmacro %}

內容解密：

  • get_column_values()巨集接收column_name和table_name作為引數，動態生成查詢以傳回指定列的不同值。
  • 我們實作了兩個單獨的巨集呼叫，分別用於檢索不同的payment_types和payment_methods。

自動過濾非佈署環境的資料集

另一個實用的巨集範例是根據環境自動過濾資料集。為此，我們可以在macros目錄中建立一個名為limit_dataset_if_not_deploy_env.sql的新巨集，並將範例5-16中的程式碼複製進去。

範例5-16：limit_dataset_if_not_deploy_env巨集

{# Macro that considering the target name, limits the amount of data queried for the nbr_months_of_data defined #}
{% macro limit_dataset_if_not_deploy_env(column_name, nbr_months_of_data) %}
    -- limit the amount of data queried if not in the deploy environment.
    {% if target.name != 'deploy' %}
        where {{ column_name }} > DATE_SUB(CURRENT_DATE(), INTERVAL {{ nbr_months_of_data }} MONTH)
    {% endif %}
{% endmacro %}

然後，在fct_orders模型中，在左連線之後加入範例5-17中的程式碼。

範例5-17：在fct_orders中呼叫limit_dataset_if_not_deploy_env巨集

with orders as (
    select * from {{ ref('stg_jaffle_shop_orders' )}}
),
payment_type_orders as (
    select * from {{ ref('int_payment_type_amount_per_order' )}}
)
select
    ord.order_id,
    ord.customer_id,

內容解密：

  • 此巨集根據目標環境名稱決定是否限制查詢的資料量。
  • 如果不是佈署環境，則根據指定的月份數過濾資料。

dbt 進階應用：巨集與套件管理

在 dbt 專案中，巨集（Macros）與套件（Packages）是兩個強大的功能，能夠大幅提升資料模型的開發效率與可維護性。本文將探討這兩個主題，並介紹如何在實際專案中應用它們。

使用 SQL 巨集提升程式碼重用性

dbt 的巨集功能允許開發者使用 Jinja 語法建立可重用的程式碼區塊。這些巨集可以被用來自動化常見的資料處理任務，或者根據不同的環境調整資料模型的行為。

實作範例：限制資料集範圍

ord.order_date,
pto.cash_amount,
pto.credit_amount,
pto.total_amount,
case
when status = 'completed'
then 1
else 0
end as is_order_completed
from orders as ord
left join payment_type_orders as pto ON ord.order_id = pto.order_id
-- 新增巨集以限制資料範圍
{{- limit_dataset_if_not_deploy_env('order_date', 3) }}

內容解密：

  1. limit_dataset_if_not_deploy_env 巨集用於根據執行環境調整資料範圍。
  2. 在非佈署環境中，自動限制 order_date 在最近 3 個月內。
  3. 這種做法能夠大幅減少開發環境中的資料處理量。

自訂 dbt 核心巨集

dbt 允許開發者自訂核心巨集，以滿足特定專案需求。例如，generate_schema_name 巨集可以用來產生符合專案命名慣例的 Schema 名稱。

dbt 套件管理

套件是 dbt 用來組織和共用程式碼與資源的有效方式。透過套件，開發者可以：

  • 將專案模組化，提高可維護性
  • 重用程式碼與資源，減少重複工作
  • 與社群共用成果，促進合作

安裝套件步驟

  1. 建立 packages.yml 檔案（若尚未存在）
  2. 在 packages.yml 中新增所需的套件組態
  3. 執行 dbt deps 命令安裝套件
  4. 使用 dbt test 驗證套件功能

範例：安裝 dbt_utils 套件

packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1

執行 dbt deps 後，若安裝成功將顯示成功訊息。

深入探索dbt套件與語義層

處理套件相容性問題

當遇到套件與dbt版本不相容的問題時，首先需確認目前使用的dbt版本是否支援該套件。檢查套件的官方檔案或儲存函式庫可取得相容的dbt版本資訊。另一解決方案是更新套件版本，使其與目前的dbt版本相容。若仍有問題，可尋找提供類別似功能且相容的替代套件。

探索dbt_utils套件

dbt_utils是由dbt Labs開發和維護的套件，提供了多種實用的函式、巨集和其他資源，能有效擴充套件和增強dbt專案的功能性。該套件包含：

  • 用於執行常見任務的輔助函式，如生成欄位列表、格式化日期和時間戳，以及處理空值
  • 用於更具表達性和彈性地表示資料的自訂資料型別，如陣列、範圍和間隔
  • 用於除錯和測試dbt專案的工具，如日誌記錄函式和測試框架
  • 用於執行各種任務的巨集和模型，如資料操作、視覺化和測試

在巨集和模型中使用套件

在dbt專案中，安裝套件後可透過特定的語法呼叫其巨集和模型。例如，使用dbt_utils.date_spine()巨集生成一系列日期：

{{ dbt_utils.date_spine(
    datepart="day",
    start_date="cast('2023-01-01' as date)",
    end_date="cast('2023-02-01' as date)"
) }}

內容解密：

此巨集用於生成指定日期範圍內的日期序列。在此例中，它生成2023年1月1日至2023年2月1日（不含）之間的每日日期。datepart引數指定了日期的粒度。

另一個例子是使用dbt_utils.safe_divide()巨集進行安全的除法運算：

select
    order_id,
    customer_id,
    cash_amount,
    total_amount,
    {{ dbt_utils.safe_divide('cash_amount', 'total_amount') }}
from {{ ref('fct_orders') }}

內容解密：

此巨集避免了除以零或空值的錯誤，確保了運算的穩定性。當分母為零或空值時，它傳回空值而非丟擲錯誤。

dbt語義層

dbt語義層是資料分析和商業智慧中的關鍵元件，它在原始資料和有意義的洞察之間架起橋樑。透過引入實體的概念，dbt語義層能夠自動推斷資料之間的連線，從而簡化了資料結構並促進了組織內對資料的共同理解。

語義層的核心元件

語義層主要由語義模型和度量指標組成。語義模型是基礎構建塊，包含三個主要元素：實體、維度和度量指標。這種設計使得資料團隊能夠更容易地管理和演進他們的資料模型。

實體與關係的自動推斷

透過定義實體及其關係，dbt語義層能夠自動推斷出資料圖中的邊緣，從而減少了手動維護邏輯的需求。這種方法封裝了語義邏輯，促進了更廣泛的度量和維度組合，並產生了更乾淨的SQL。

玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。