【Dataform】pre_operationsを使ってdelete insertを実現する方法

Dataform では incremental モードを使ってデータを積み上げていくことができます。 しかし、データを積み上げていくことのみしか出来ず、仮に一度実行した処理を再度実行すると同じデータがシンク先に書き込まれます。 要はDELETE INSERT には対応していないということになります。

現時点の Dataform の機能としては対応していないということで工夫が必要になり、それを実現するためにpre_operationsを使います。

pre_operations

pre_operationsは SQLX ファイル内に記述し、テーブルの作成前に実行できる SQL ステートメントです。 Dataform の docs の例 では、一時的なファンクションを作っています。

  pre_operations {
    CREATE TEMP FUNCTION AddFourAndDivide(x INT64, y INT64)
      RETURNS FLOAT64
      AS ((x + 4) / y);
  }

ref: https://cloud.google.com/dataform/docs/table-settings?hl=ja#execute-sql-before-table

ちなみにテーブルの作成後に実行できる SQL ステートメントpost_operationsも存在します。

pre_operationsの中に DELETE 文を記述することで、同じデータが追加されず冪等性のある処理にすることが出来ます。

次は、WITH のサブクエリ内で作ったソースデータを、sample_table に入れている処理です。 pre_operations の中で、2024-05-13 の日付のデータを DELETE する処理を実行しています。 その後、WITH のサブクエリのデータを sink先のtable に insert しています。

ソース元は実際のテーブルへと、また日付は動的に変更できるようにする必要はありますが、これで date 単位でのDELETE INSERTの洗替処理は実行できます。

config {
  type: "incremental",
  schema: "sample_dataset",
  name: "sink_table",
}

pre_operations {
    DELETE FROM ${self()} WHERE date = '2024-05-13'
}

WITH source_table AS (
  SELECT
    DATE('2024-05-13') AS date,
    1 as user_id,
    100 as value,
  UNION ALL
  SELECT
    DATE('2024-05-13') AS date,
    2 as user_id,
    200 as value,
)

SELECT
  date,
  user_id,
  value
FROM source_table
WHERE date = '2024-05-13'

Dataform の公式 GitHub の issue にて質問中...

返事が返ってきませんが、公式 GitHub の issue にて質問をしています。

github.com

現状色々調べた中で pre_operations を使って DELETE INSERT の洗替処理をする方法しかなさそうというところに行き着きました。 公式が提供する機能で同じ処理と結果が得られるのであれば、そちらのほうが嬉しいので実現を待ちたいです。

英語の翻訳はChatGPTがよしなにやってくれるので便利ですね。