こちらの続編です。

追加の要件

取り込む対象のディレクトリには過去のデータが格納されている。
DLTパイプラインを初回実行する際には、過去のデータは取り込みたくない。
以降のDLTパイプライン実行では、初回パイプライン実行後に追加されたファイルのみを用いて、洗い替えを行いたい。

DLTパイプラインの設定

DLTパイプラインを初回に実行する際、既存のファイルを処理するかどうかを指定するオプションがあります。

cloudFiles.includeExistingFiles:

ストリーム処理入力パスに既存のファイルを含めるか、初期セットアップ後に到着する新しいファイルのみを処理するかどうか。このオプションは、初めてストリームを開始するときにのみ評価されます。ストリームの再開後にこのオプションを変更しても効果はありません。

こちらの設定をパイプラインを実装するノートブックで追加します。

dlt_complete_sql

CREATE
OR REFRESH STREAMING TABLE complete_bronze_w_filter TBLPROPERTIES (
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5',
  'delta.columnMapping.mode' = 'name',
  'pipelines.reset.allowed' = 'false'
) COMMENT "Parquetをそのまま保持するブロンズテーブル" AS
SELECT
  *,
  current_timestamp() as processed -- 処理時刻
FROM
  cloud_files(
    "/tmp/takaaki.yayoi@databricks.com/dlt/landing/",
    "parquet",
    map("cloudFiles.includeExistingFiles", "false") -- 初回実行時に既存ファイルを無視
  )

今回はシンプルにするため、シルバーテーブルは作成せずにブロンズテーブルのみとします。

DLTパイプラインの作成

上述のノートブックを指定してDLTパイプラインを作成します。前回同様、ターゲットはUCのtakaakiyayoi_catalog.dltにします。

削除用クエリーの作成

こちらも前回同様です。

table_reset

DELETE FROM takaakiyayoi_catalog.dlt.complete_bronze_w_filter;

テーブル確認用クエリーの作成

こちらは新たなコンポーネントです。これは、初回のジョブ実行時にはレコードを削除するテーブルが存在せず、その際にDELETE FROMを実行するとエラーになるのを避けるためです。

check_table

SELECT * FROM takaakiyayoi_catalog.dlt.complete_bronze_w_filter LIMIT 1;

ジョブの作成

上述のコンポーネントをタスクとして組み合わせます。

タスク: check_table

種類: SQL
SQLタスク: クエリー
SQLクエリー: check_table

タスク: incremental_load_w_filter_wo_reset

種類: Delta Live Tablesパイプライン
パイプライン: sql_dlt_complete_w_filter
Delta Live Tablesパイプラインで完全なリフレッシュをトリガーする: チェックしない
依存先: check_table
次の場合に実行: すべて失敗しました

タスク: reset_target_table

種類: SQL
SQLタスク: クエリー
SQLクエリー: reset_table
依存先: check_table
次の場合に実行: すべて成功しました

タスク: incremental_load_w_filter_w_reset

種類: Delta Live Tablesパイプライン
パイプライン: sql_dlt_complete_w_filter
Delta Live Tablesパイプラインで完全なリフレッシュをトリガーする: チェックしない
依存先: reset_target_table
次の場合に実行: すべて成功しました

ジョブ定義のJSONはこちらとなります。

{
    "run_as": {
        "user_name": "takaaki.yayoi@databricks.com"
    },
    "name": "increment_load_complete_update_w_filter",
    "email_notifications": {
        "no_alert_for_skipped_runs": false
    },
    "webhook_notifications": {},
    "timeout_seconds": 0,
    "max_concurrent_runs": 1,
    "tasks": [
        {
            "task_key": "check_table",
            "run_if": "ALL_SUCCESS",
            "sql_task": {
                "query": {
                    "query_id": "bb7bfe8f-166c-42e6-8165-3266aa7d5d91"
                },
                "warehouse_id": "2e1a7a349827f40c"
            },
            "timeout_seconds": 0,
            "email_notifications": {},
            "notification_settings": {
                "no_alert_for_skipped_runs": false,
                "no_alert_for_canceled_runs": false,
                "alert_on_last_attempt": false
            }
        },
        {
            "task_key": "incremental_load_w_filter_wo_reset",
            "depends_on": [
                {
                    "task_key": "check_table"
                }
            ],
            "run_if": "ALL_FAILED",
            "pipeline_task": {
                "pipeline_id": "b05bea64-463a-4e9d-b3a0-c41dde5aa104",
                "full_refresh": false
            },
            "timeout_seconds": 0,
            "email_notifications": {},
            "notification_settings": {
                "no_alert_for_skipped_runs": false,
                "no_alert_for_canceled_runs": false,
                "alert_on_last_attempt": false
            }
        },
        {
            "task_key": "reset_target_table",
            "depends_on": [
                {
                    "task_key": "check_table"
                }
            ],
            "run_if": "ALL_SUCCESS",
            "sql_task": {
                "query": {
                    "query_id": "ba8c07e3-7f5d-4bb8-a614-4484afd90830"
                },
                "warehouse_id": "2e1a7a349827f40c"
            },
            "timeout_seconds": 0,
            "email_notifications": {},
            "notification_settings": {
                "no_alert_for_skipped_runs": false,
                "no_alert_for_canceled_runs": false,
                "alert_on_last_attempt": false
            }
        },
        {
            "task_key": "incremental_load_w_filter",
            "depends_on": [
                {
                    "task_key": "reset_target_table"
                }
            ],
            "run_if": "ALL_SUCCESS",
            "pipeline_task": {
                "pipeline_id": "b05bea64-463a-4e9d-b3a0-c41dde5aa104",
                "full_refresh": false
            },
            "timeout_seconds": 0,
            "email_notifications": {},
            "notification_settings": {
                "no_alert_for_skipped_runs": false,
                "no_alert_for_canceled_runs": false,
                "alert_on_last_attempt": false
            }
        }
    ],
    "format": "MULTI_TASK"
}

動作確認

データを準備します。別にノートブックを作成し、7/6のデータのみを保存します。

data_07_06 = [{"Date": "2023/07/06", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        {"Date": "2023/07/06", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        {"Date": "2023/07/06", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": None},
        {"Date": "2023/07/06", "Category": 'D', "ID": 4, "Value": 33.87, "Truth": True}
        ]

df_07_06 = spark.createDataFrame(data_07_06)
display(df_07_06)

data_07_07 = [{"Date": "2023/07/07", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        {"Date": "2023/07/07", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        #{"Date": "2023/07/07", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": None},
        {"Date": "2023/07/07", "Category": 'D', "ID": 4, "Value": 33.87, "Truth": True}
        ]

df_07_07 = spark.createDataFrame(data_07_07)
display(df_07_07)

data_07_08 = [#{"Date": "2023/07/08", "Category": 'A', "ID": 1, "Value": 121.44, "Truth": True},
        #{"Date": "2023/07/08", "Category": 'B', "ID": 2, "Value": 300.01, "Truth": False},
        {"Date": "2023/07/08", "Category": 'C', "ID": 3, "Value": 10.99, "Truth": True},
        {"Date": "2023/07/08", "Category": 'E', "ID": 5, "Value": 33.87, "Truth": True}
        ]

df_07_08 = spark.createDataFrame(data_07_08)
display(df_07_08)

save_path = f"{data_path}/2023/07/06/"
df_07_06.write.format("parquet").option("header", "true").mode("overwrite").save(save_path)

display(dbutils.fs.ls(f"{data_path}/2023/07"))

初回実行

ジョブを実行します。

初回はテーブルが存在しないので、上のルートが実行されます。テーブルをリセットせずにDLTパイプラインが実行されます。

DLTパイプラインの実行結果を確認すると、cloudFiles.includeExistingFiles=falseが設定されているので、7/6のデータは処理されません。

テーブルは作成されますが、中身は空です。

2回目以降の実行

別のノートブックで7/7のデータを保存します。

save_path = f"{data_path}/2023/07/07/"
df_07_07.write.format("parquet").option("header", "true").mode("overwrite").save(save_path)

display(dbutils.fs.ls(f"{data_path}/2023/07"))

7/6と7/7のデータが存在しています。

再度ジョブを実行します。今度はテーブルが存在しているので下のルートが実行されます。

今度は7/7のデータのみが処理され、テーブルが洗い替えされます。

7/8のデータが追加された後でジョブを実行すると7/8のデータで洗い替えされます。

Databricksクイックスタートガイド

Databricksクイックスタートガイド

Databricks無料トライアル

Databricks無料トライアル