DataForm で変換ワークフローを定義する SQLX を徹底解説

Google Cloud のデータ変換サービス DataForm 入門 では、DataForm の概要と DataForm CLI を使ったプロジェクトの新規作成、変換ワークフローの実行について解説しました。

本記事では、DataForm で変換ワークフローを定義するための SQLX ファイルの書き方について解説します。

SQLX とは

DataForm で変換ワークフローを定義するためのファイル形式が SQLX です。

SQLX ファイルは SQL ファイルの拡張で、SQL ファイル自体も 有効な SQLX ファイルです。SQLX ファイルは SQL で書けるクエリに加えて、開発をより高速にスケーラブルにするための機能が加えられています。

具体的には、 config ブロックでデータソースの宣言やテストの定義をしたり、body ブロックで JavaScript で変数や関数を定義してクエリ内で利用したりすることができます。

SQLX ファイルの書き方

SQLX ファイルは、以下のような構成で書くことができます。

• config ブロック
• body ブロック
• pre_operations ブロック (オプション)
• post_operations ブロック (オプション)

ここからは実際にテーブルを作成する例とともに各ブロックの書き方を解説します。

config ブロック

config ブロックでは、データソースの宣言やテストの定義を行います。

config {
  type: "view",
  materialized: true,
  database: "{project_id}",
  schema: "{dataset_name}",
  name: "{table_name}",
  description: "This is a sample view.",
  dependencies: [ "sqlx_filename1", "sqlx_filename2" ],
}

以下に普段使うことが多い config ブロックの設定項目を紹介します。

• type : データソースの種類を指定します。
  • table : テーブル
  • view : ビュー
  • incremental : 増分テーブル
• materialized : type が view である場合に、実体化されたビューを作成するか否かを指定します。
  • boolean 型で true または false を指定します。true の場合は実体化されたビューを作成します。
• database : データベース名を指定します。BigQuery の場合はプロジェクトIDです。
• schema : スキーマ名を指定します。BigQuery の場合はデータセット名です。
• name : データソース名を指定します。BigQuery の場合はテーブル名やビュー名です。
• description : データソースの説明を指定します。
• dependencies : SELECT ステートメントで参照されていないものの、 そのSQLXファイルが定義する変換が実行される前に用意されておく必要があるテーブルを宣言します。 SELCT ステートメントで直接参照されるテーブルを記述する必要はありません。
  • 例: dependencies: [ "sqlx_filename1", "sqlx_filename2" ]
• bigquery : BigQuery 固有の設定を指定します。
  • partitionBy : パーティションテーブルを分割するキーを指定します。例: タイムスタンプや日付など。
  • updatePartitionFilter : パーティションテーブルの更新フィルタを指定します。例: date_column > CURRENT_DATE()
• assertions : 作成するテーブルやビューに対するアサーションを定義できます。 テーブルやビューが期待通りのデータを保持しているかを確認するテストクエリを柔軟に定義することができます。後述します。

config ブロックのその他項目については、 DataForm コアのリファレンス を参照してください。

body ブロック

body ブロックは、BigQuery がサポートする SQL ステートメントや JavaScript を用いて複雑な変換ワークフローを定義するコードブロックです。

単純な変換であれば、SELECT ステートメントを記述するだけで良いですが、 複雑な変換を行う場合やプロジェクトで共通の処理をモジュール化して再利用したい場合に、 JavaScript を用いて変数や関数を定義して利用することができます。

なお、ここでは SQLX に特有の構文について解説し SQL の書き方について特段触れません。 SQL の書き方については、BigQuery のドキュメント を参照してください。

テーブル・ビューを定義する

CREATE ステートメントではなく SELECT ステートメントでテーブルを定義/作成します。

SQLX ファイルで以下のように書くと、 実行時に CREATE ステートメントに変換されテーブルが作成されます。ビューも同様です。

definitions/outputs/user_purchase_history.sqlx

config {
  type: "table",
}

WITH user_id AS (
  SELECT
    id AS user_id,
    age AS user_age,
    created_at AS user_created_at,
  FROM ${ref("raw_user_table")}
), 

purchase_history AS (
  SELECT
    id AS purchase_id,
    user_id AS user_id,
    item AS purchase_item,
    price AS purchase_price,
  FROM ${ref("raw_purchase_table")}
)

SELECT
  purchase_id,
  user_id,
  purchase_item,
  purchase_price,
  user_created_at,
  user_age
FROM user_id
LEFT JOIN purchase_history
USING(user_id)

増分テーブル（インクリメンタルテーブル）を定義する

config ブロックの type を incremental に指定することで、増分テーブルを定義することができます。 次の例では、パーティション分割テーブルである raw_purchase_table を参照して パーティション分割された増分テーブル user_purchase_history を作成・更新する SQLX ファイルを示しています。

definitions/outputs/incremental_user_purchase_history.sqlx

config {
  type: "incremental",
  uniqueKey: ["purchase_id"],
  bigquery: {
    partitionBy: "DATE(event_timestamp)",
    updatePartitionFilter:
        "event_timestamp >= timestamp_sub(current_timestamp(), interval 24 hour)"
  }
}

pre_operations {
  -- pre_operations ブロックでパーティション・プルーニングの条件に用いる
  -- timestamp などを定義しておくと body ブロックでは定数として用いることができる。
  DECLARE event_timestamp_checkpoint DEFAULT (
    ${when(incremental(),
    `SELECT max(event_timestamp) - INTERVAL 2 DAY FROM ${self()}`,
    `SELECT timestamp("2000-01-01")`)}
  )
}

WITH user_id AS (
  SELECT
    id AS user_id,
    age AS user_age,
    created_at AS user_created_at,
  FROM ${ref("raw_user_table")}
),

purchase_history AS (
  SELECT
    id AS purchase_id,
    user_id AS user_id,
    item AS purchase_item,
    price AS purchase_price,
    event_timestamp
  FROM ${ref("raw_purchase_table")}
  -- 以下は定数式となり、きちんとパーティション・プルーニングが働く。
  WHERE event_timestamp > event_timestamp_checkpoint
)

SELECT
  purchase_id,
  user_id,
  purchase_item,
  purchase_price,
  user_created_at,
  user_age
FROM
  user_id
LEFT JOIN purchase_history
USING(user_id)

パーティション分割テーブルを参照して増分テーブルを更新する際は、テーブルの更新時に参照先をフルスキャンさせないことが重要です。 上に示した例では、pre_operations ブロックで event_timestamp_checkpoint を定数として定義し、 purchase_history テーブルで event_timestamp が event_timestamp_checkpoint より新しいパーティションのみを参照するようにしています。

また purchase_id をユニークキーに指定して、更新する増分テーブルに重複がないようにしています。 incremental_user_purchase_history テーブルと一致する purchase_id が存在しない場合はインサート、一致する purchase_id がある場合はアップデートされます。

bigquery: {
    partitionBy: "DATE(event_timestamp)",
    updatePartitionFilter:
        "event_timestamp >= timestamp_sub(current_timestamp(), interval 24 hour)"
  }

さらにこの部分では updatePartitionFilter を指定して一致している行を探索するパーティションを限定しています。

上記の例のように、パーティション分割テーブルを参照して増分テーブルを作成する際は pre_operations を用いて パーティション・プルーニングが可能です。またユニークキーを指定してテーブルをマージさせる際は、 一致する行をテーブル全体で探索させないように updatePartitionFilter を用いることができます。

pre_operations, post_operations ブロック

pre_operations, post_operations ブロックにカスタム SQL を書くことで、テーブルの作成の前後で何らかの処理を行わせることができます。

pre_operations ブロックはパーティション分割テーブルを参照して増分テーブルを作成する方法で既に登場しました。 そのほかにもテーブル作成の前に行いたい処理はここで定義すると良いです。

post_operations ブロックに書いたカスタムSQLステートメントは、 body ブロックのテーブル作成の後に実行されます。

post_operations {
  -- テーブル作成後に行いたい処理を記述する。
}

post_operations ブロックはオプションですが、作成したテーブルに対してユーザーの権限を設定したい場合などに有用です。

アサーションについて

テーブルが期待通りのデータを保持しているかを確認するために、config ブロックの assertions にアサーションを定義することができます。

書き方は次の２種類があります。

1. config ブロックに assertions を記述する方法
2. 別のアサーション専用 SQLX ファイルを作成する方法

1 つ目の方法は、config ブロックにアサーションを記述する方法です。以下の例では、 作成する user_purchase_history テーブルに対して user_id が NULL である行が存在しないことをアサーションしています。

definitions/outputs/user_purchase_history.sqlx

config {
  type: "table",
  assertions: {
    rowConditions: [
      `user_id is NULL`
    ]
  }
}

別の SQLX ファイルを作成する方法で、同様のアサーションを次のように定義してみます。 SQLX ファイルで作成する際は、アサーションが成功した場合に行を返さないクエリを書く必要があります。 行が返されるとアサーションは失敗します。

definitions/assertions/assert_user_purchase_history.sqlx

config {
  type: "assertion",
}

SELECT *
FROM ${ref("user_purchase_history")}
WHERE user_id IS NULL

DataForm では SQL ワークフローを更新するたびにアサーションが実行され、 アサーションが失敗した場合はアラートを受け取ることができます。 また、BigQuery に dataform_assertions.assertion_name の形式でビューが作成され、 アサーションの結果を確認することもできます。

より詳細なアサーションの書き方については、 アサーションを使用したテーブルのテスト | DataForm ドキュメント を参照してください。

まとめ

本記事では、DataForm で変換ワークフローを定義するための SQLX ファイルの書き方について解説しました。 config ブロックで SQLX ファイルの設定や単体テストのためのアサーションの定義を行い、 body ブロックで BigQuery がサポートする SQL ステートメントや JavaScript を用いて複雑な変換ワークフローを定義することができます。

DataForm に関する次の記事では、DataForm プロジェクトと GitHub リポジトリを接続して、CI/CD する方法について解説します。

参考

• DataFormのドキュメント | Google Cloud
• DataForm コアのリファレンス

備考

Hakky ではエンジニアを募集中です！まずは話してみたいなどでも構いませんので、ぜひお気軽に採用ページからお問い合わせくださいませ。