【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'