Open6

【dbt】dbt jinja

dbt
YuichiYuichi

for 構文

https://dev.classmethod.jp/articles/dbt-jinja/

{% set source_columns = ["Field_P_Id","Field_P_Name", "Field_P_Alias", "Field_P_Type", "Field_P_Required", "Field_P_Max", "Field_P_Min", "Field_P_DecimalFraction", "Field_P_ReferTo", "Field_P_ReferTo", "Field_P_ReferVia","time"] %}

SELECT
  {{ source_columns[0]  ~ " AS " ~ source_columns[0]  | replace("_P", "") | lower }}
  {% for col in source_columns[1:] %}
      {% if col == 'time' %}
        {{ ",MAX(time) AS insert_at" }}
      {% else %}
        {{ ",MAX_BY(" ~ col ~ ",time) AS " ~ col | replace("_P", "") | lower }}
      {% endif %}
  {% endfor %}
FROM {{ source('porters', 'field') }}
GROUP BY field_id

↓↓↓↓↓↓↓↓

SELECT
    Field_P_Id AS field_id
    ,MAX_BY(Field_P_Name,time) AS field_name
    ,MAX_BY(Field_P_Alias,time) AS field_alias
    ,MAX_BY(Field_P_Type,time) AS field_type
    ,MAX_BY(Field_P_Required,time) AS field_required
    ,MAX_BY(Field_P_Max,time) AS field_max
    ,MAX_BY(Field_P_Min,time) AS field_min
    ,MAX_BY(Field_P_DecimalFraction,time) AS field_decimalfraction
    ,MAX_BY(Field_P_ReferTo,time) AS field_referto
    ,MAX_BY(Field_P_ReferTo,time) AS field_referto
    ,MAX_BY(Field_P_ReferVia,time) AS field_refervia
    ,MAX(time) AS insert_at
FROM `project_id`.`porters`.`field`
GROUP BY field_id
YuichiYuichi
{% set action_months = ["2024-05-01", "2024-06-01", "2024-07-01"] %}

{% for month in action_months %}
SELECT DATE_TRUNC('{{ month }}',MONTH) AS action_month
{% if not loop.last %}
UNION ALL
{% endif %}
{% endfor %}
YuichiYuichi

実際のデータセット名とテーブル名を取得する方法

dbt では、ソーステーブルの実際の BigQuery の データセット名とテーブル名 を adapter.get_relation を使って取得できます。

方法 1: adapter.get_relation を使う

以下のように Jinja マクロを使うことで、実際のテーブル情報を取得できます。

{% set relation = adapter.get_relation(database=target.database, schema="ads", identifier="google") %}

SELECT 
    '{{ relation.database }}' AS project_id,
    '{{ relation.schema }}' AS dataset_name,
    '{{ relation.identifier }}' AS table_name

これにより、実際の project_id, dataset_name, table_name を SQL の SELECT 文として取得できます。

方法 2: dbt の内部変数 source を利用

もし source() を直接 SQL で使う場合、以下のように {% set %} を使って値を取得できます。

{% set relation = source("ads", "google") %}

SELECT 
    '{{ relation.database }}' AS project_id,
    '{{ relation.schema }}' AS dataset_name,
    '{{ relation.identifier }}' AS table_name

方法 3: this を使う

モデル内で source() を使う場合、dbt では this 変数を使うこともできます。

SELECT
    '{{ this.database }}' AS project_id,
    '{{ this.schema }}' AS dataset_name,
    '{{ this.identifier }}' AS table_name
FROM {{ source("ads", "google") }}

この方法を使うと、dbt の実行時に ads.google の実際の プロジェクトID、データセット名、テーブル名 を取得できます。

YuichiYuichi

クエリ書く時、クラスタとかパーティション確認したい時

このマクロ get_table_columns は、指定されたテーブルのカラム情報を取得し、カラムの順序、名前、説明、データ型、パーティション列、クラスタリング列の情報を返します。使い方として、source や ref を指定することで、任意のモデルやソースからカラム情報を取得することができます。

{% macro get_table_columns(model) %}
    select
        c.ordinal_position as column_no,
        c.column_name,
        d.description,
        c.data_type,
        c.is_partitioning_column as partition_by,
        c.clustering_ordinal_position as cluster_by
            from `{{ model.schema }}`.INFORMATION_SCHEMA.COLUMNS as c -- fmt: off
            left join
                `{{ model.schema }}`.INFORMATION_SCHEMA.COLUMN_FIELD_PATHS  as d -- fmt: on
        on c.column_name = d.column_name
        and c.table_name = d.table_name
    where c.table_name = '{{ model.identifier }}'
    order by c.ordinal_position
{% endmacro %}

{# 
サンプル
{{ get_table_columns(source("jaffle_shop", "customers")) }}
{{ get_table_columns(ref("stg_customers")) }}
#}
YuichiYuichi

DWH整備する時

get_column_samples マクロの実装は、指定されたテーブルの各カラムからサンプル値を取得し、各カラムに関連するメタデータを一緒に表示するものです。このマクロの構造は良好ですが、いくつかの改善点と補足を加えることができます。

{% macro get_column_samples(model, sample_size=10) %}

{% set columns = dbt_utils.get_filtered_columns_in_relation(from=model) %}

with
    column_info as (
        select
            c.ordinal_position as column_no, c.column_name, d.description, c.data_type,
        from `{{ model.schema }}`.INFORMATION_SCHEMA.COLUMNS c -- fmt: off
        left join
            `{{ model.schema }}`.INFORMATION_SCHEMA.COLUMN_FIELD_PATHS d -- fmt: on
            on c.column_name = d.column_name
            and c.table_name = d.table_name
        where c.table_name = '{{ model.identifier }}'
    ),
    sample_values as (
        {% for col in columns %}
            select '{{ col }}' as column_name, cast({{ col }} as string) as value
            from (select {{ col }} from {{ model }} where {{ col }} is not null limit {{ sample_size }})
            {% if not loop.last %}
                union all
            {% endif %}
        {% endfor %}
    )
select
    ci.column_no,
    ci.column_name,
    ci.description,
    ci.data_type,
    string_agg(distinct sv.value, " | ") as samples
from column_info ci
left join sample_values sv on ci.column_name = sv.column_name
group by ci.column_no, ci.column_name, ci.description, ci.data_type
order by ci.column_no asc

{% endmacro %}
YuichiYuichi

https://belonginc.dev/members/shuhei/posts/dbt-utils-generate-surrogate-key

generate_surrogate_key とは？

generate_surrogate_key マクロは、その名の通りサロゲートキーを作成してくれるマクロです。

NULLABLE な列を含む列であっても安全にサロゲートキーを作成してくれるので、利用する側としてとても便利なマクロだと思いました。NULL のケアをマクロ側で行ってくれるので、こちら側が coalesce しなくて済むのはスマートですね！

dbt project の変数 surrogate_key_treat_nulls_as_empty_strings を False に設定することで NULL 置換文字を空文字とすることもできますが、これは generate_surrogate_key マクロの前身である surrogate_key マクロが NULL を空文字で扱っていたことに由来するそうです。

{{ dbt_utils.generate_surrogate_key(['field_a', 'field_b'[,...]]) }}