概要
Databricks DLT の APPLY CHANGES API について仕様を検証した結果を共有します。
結論として、Key 列が NULL のレコードも連携されるため、実運用時には Key 列が NULL のレコードを除外するなどの対策が必要になりそうです。
今回の主な検証結果は下記のとおりです。
- Key 列が NULL の場合でもエラーにならず連携される
- ソース側で key と sequence_by が重複していてもエラーにならない
検証パターンと結果
以下の表に各検証パターンとその結果をまとめました。
| # | 検証パターン | 結果 |
|---|---|---|
| 1 | ソースで Key と sequence_by が重複している場合の UPDATE 検証 | エラーにならず、ID=2 のレコードが反映された |
| 2 | ソースで Key と sequence_by が重複している場合の INSERT 検証 | エラーにならず、ID=3 のレコードが反映された |
| 3 | ソースで sequence_by が NULL の場合の動作検証 | エラーにならず、ID=4 のレコードが追加された |
| 4 | ソースで sequence_by に値を設定した場合の動作検証 | ID=4 のレコードが更新された |
| 5 | ソースで sequence_by を再び NULL で追加した場合の動作検証 | ID=4 のレコードは変更されず、前回のまま維持された |
| 6 | ソースで Key が NULL の場合の動作検証 | NULL の ID レコードが追加された |
| 7 | ソースで Key が NULL のレコードを追加した場合の動作検証 | NULL の ID レコードが更新された |
検証コードと結果
事前準備
カタログを設定します。
%sql
USE CATALOG manabian_test_01;
ソースのスキーマとテーブルを作成します。
%sql
CREATE SCHEMA IF NOT EXISTS dlt_source_01;
%sql
CREATE OR REPLACE TABLE dlt_source_01.table_01
AS
SELECT
1 AS id,
'John' AS string_col,
CAST('2025-01-01' AS TIMESTAMP) AS ingest_ts
;
SELECT * FROM dlt_source_01.table_01;
続いて、ソーステーブルを参照する DLT の処理を記述して実行します。
DLT の書き込み先スキーマは dlt_test_01 です。
from dlt import *
from pyspark.sql.functions import *
catalog = "manabian_test_01"
schema = "dlt_test_01"
# table_01 の処理
table_01 = "table_01"
ftable_01 = f"{catalog}.{schema}.{table_01}"
@dlt.table(
name=ftable_01,
)
def customers_bronze_ingest_flow():
return spark.readStream.table(f"{catalog}.dlt_source_01.table_01")
# table_02 の処理
table_02 = "table_02"
ftable_02 = f"{catalog}.{schema}.{table_02}"
dlt.create_streaming_table(
name=ftable_02,
)
dlt.apply_changes(
target=ftable_02,
source=ftable_01,
keys=["id"],
sequence_by=col("ingest_ts"),
stored_as_scd_type=1,
)
初回実行後、結果が想定通りであることを確認します。
%sql
SELECT * FROM dlt_test_01.table_02
次に、ソーステーブルにデータを書き込んで DLT の処理を再度実行し、結果を確認します。
%sql
WITH src AS (
SELECT
1 AS id,
'John++++++++++' AS string_col,
CAST('2025-02-01' AS TIMESTAMP) AS ingest_ts
UNION ALL
SELECT
2 AS id,
'Doe' AS string_col,
CAST('2025-02-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02
1. ソースにて Key と sequence_by が重複している場合の UPDATE の検証
エラーにならず、ID=2 のレコードが反映されることを確認しました。
%sql
WITH src AS (
SELECT
2 AS id,
'Doe+1' AS string_col,
CAST('2025-03-01' AS TIMESTAMP) AS ingest_ts
UNION ALL
SELECT
2 AS id,
'Doe+2' AS string_col,
CAST('2025-03-01' AS TIMESTAMP) AS ingest_ts
UNION ALL
SELECT
2 AS id,
'Doe+3' AS string_col,
CAST('2025-03-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02
2. ソースにて Key と sequence_by が重複している場合の INSERT の検証
エラーにならず、ID=3 のレコードが反映されることを確認しました。
%sql
WITH src AS (
SELECT
3 AS id,
'Cat+1' AS string_col,
CAST('2025-04-01' AS TIMESTAMP) AS ingest_ts
UNION ALL
SELECT
3 AS id,
'Cat+2' AS string_col,
CAST('2025-04-01' AS TIMESTAMP) AS ingest_ts
UNION ALL
SELECT
3 AS id,
'Cat+3' AS string_col,
CAST('2025-04-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02
3. ソースにて sequence_by が NULL である場合の動作検証
エラーにならず、ID=4 のレコードが新規追加されていることを確認しました。
%sql
WITH src AS (
SELECT
4 AS id,
'Donkey' AS string_col,
NULL AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02 ORDER BY ID
4. ソースにて sequence_by に値を設定した場合の動作の検証
ID=4 のレコードが更新されたことを確認しました。
%sql
WITH src AS (
SELECT
4 AS id,
'Donkey+1' AS string_col,
CAST('2025-05-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02 ORDER BY ID
5. ソースにて sequence_by が再度 NULL で追加された場合の動作検証
ID=4 のレコードは更新されず、前回の状態が維持されました。
%sql
WITH src AS (
SELECT
4 AS id,
'Donkey+2' AS string_col,
NULL AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02 ORDER BY ID
6. ソースにて Key が NULL である場合の動作検証
ID=NULL のレコードが新規追加されていることを確認しました。
%sql
WITH src AS (
SELECT
NULL AS id,
'Elephant' AS string_col,
CAST('2025-05-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
%sql
SELECT * FROM dlt_test_01.table_02 ORDER BY ID
7. ソースにて Key が NULL のレコードが追加された場合の動作検証
ID=NULL のレコードが更新されていることを確認しました。
%sql
WITH src AS (
SELECT
NULL AS id,
'Elephant+1' AS string_col,
CAST('2025-06-01' AS TIMESTAMP) AS ingest_ts
)
INSERT INTO dlt_source_01.table_01
SELECT * FROM src;
SELECT * FROM dlt_source_01.table_01;
``
```sql
%sql
SELECT * FROM dlt_test_01.table_02 ORDER BY ID
