dbt run-operationを用いた、dbt cloud CI/CD対応のBigQuery UDF管理のすすめ

コミューンアドベントカレンダー・dbtアドベントカレンダーの17日目を担当します。
コミューン株式会社でデータエンジニアをしているよしけんです。

dbt run-operationを用いることで、BigQueryのUDF管理をどのように工夫して可用性の高い運用を楽に実現しているか、本日は紹介したいと思います。

想定する環境

• BigQueryを使っている
• dbt Cloud Teams版か、dbt Cloud Enterprise版を使っている

目的

BigQuery UDFを永続化させて様々なデータ処理で使い始めると、運用メンテナンスが困難です。
手作業だと、DROP FUNCTIONした後にCREATE FUNCTIONをしようとしたらエラーが起きてしまうかもしれません。その間にそれを参照するクエリは全てエラーになるでしょう。
そのときには切り戻しだけでなく、データのリカバリなどで疲弊してしまいそうです。

運用の手間やリスクをなくし、予めテストが通った物をリリースできる方法があります。
そういった処理を、下記のファイル構成で実現する事を紹介する記事です。

今回のデモを実際に動かすにあたって必要なサンプルコード一式は、下記レポジトリにて公開しています。
https://github.com/y-ken/dbt-udf-camel_to_snake_case

特徴

• リリース作業は特に不要なので、うっかり反映タイミングが遅れて障害が起きることがない
  • そのUDFを利用しているJobの最初で毎回dbt run-operationを呼ぶため
• dbt cloud CI/CD対応している
  • dbt cloudでCI/CD連携していると、Pull-Requestを作ると自動的に専用のdatasetが作られる
  • 自動的に作られたdatasetの中でUDFを作り、動作テストが行われる仕組み
• 本番環境ではdbt_prod_udfsにUDFが配置され、開発環境ではdbt_dev_udfsとなる。
  • profiles.ymlで定義したdatasetの後ろに_udfsを追加している
  • Pull-Requestの時は1つのdatasetにまとまってて欲しいので、接尾辞の追加ナシ
• UDFのリストは dbt_project.yml vars.udfsに配列で管理
• UDFがいくつ増えても、使うコマンドは2種類のみ
  • $ dbt run-operation run_operation_create_udfs
  • $ dbt run-operation run_operation_test_udfs

dbt cloudでの使い方

PR作成時に動くCI Jobや、そのUDFを使うであろうjobの最初でCDをしています。
以下のようにdbr run-operationを呼ぶだけです。

dbt run-operation run_operation_create_udfs
dbt run-operation run_operation_test_udfs
dbt list --select state:modified
dbt list --select state:modified+
dbt build --select state:modified

サンプルとして用いるUDF

UserId -> user_id のようなカラム変換を行う、camel_to_snake_case というUDFを例に説明します。
これは本番DBでは列名がCamelCaseとなっているが、DWHではsnake_caseで扱いたいときに有用なUDFです。これの詳しい活用方法は別の記事で紹介する予定です。

create or replace function `{{target.project}}`.`{{ target_dataset }}`.`camel_to_snake_case`(input STRING)
returns string
language js 
options (description="camelCaseをsnake_caseにする文字列変換UDFです")
as """
    return input.replace(/([a-z0-9])([A-Z])/g, '$1_$2')
                .replace(/([A-Z])([A-Z][a-z])/g, '$1_$2')
                .toLowerCase();
""";

run_operation_create_udfsの動作原理

dbt_project.ymlで定義したudfsというvarsでループを回して、{%- do run_query(context[udf](create_udf=true)) -%}するという工夫をしています。
テストの方も同様のループ処理となっています。

https://github.com/y-ken/dbt-udf-camel_to_snake_case/blob/main/macros/run_operation/run_operation_create_udfs.sql

UDFを追加する際のオペレーション

下記の手順でUDFの追加が行えます。
数が増えても煩雑にはならないところが気に入ってます。

1. dbt_project.ymlのvars.udfsにエントリーを追加

https://github.com/y-ken/dbt-udf-camel_to_snake_case/blob/main/dbt_project.yml

2. macros/run_operation/udf/ 配下にmacroを配置する
3. プルリクエストを作ってテストが通ればリリースできます！
   

まとめ

dbtでUDFを管理する手法は今月だけでもいくつか記事が出ており、各社流儀があるようです。
色々比較した上で、読者の方にとって一番使いやすい物を選んで頂ければと思います！
https://tech.yappli.io/entry/managing-bigquery-udf-with-dbt-macro
https://zenn.dev/mashiike/articles/44d68d98b00325