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 にて質問をしています。
現状色々調べた中で pre_operations を使って DELETE INSERT の洗替処理をする方法しかなさそうというところに行き着きました。 公式が提供する機能で同じ処理と結果が得られるのであれば、そちらのほうが嬉しいので実現を待ちたいです。
英語の翻訳はChatGPTがよしなにやってくれるので便利ですね。