マージ、挿入、追加を使用してデータを読み込む

データを変換してクレンジングしたら、パイプラインの最後の手順でターゲット テーブルにデータを読み込みます。 Azure Databricks には、 追加、 上書き、マージという 3 つの主要な読み込み方法が用意 されています。 各戦略はさまざまなユース ケースに対応し、適切な方法を選択するかどうかは、ソース データと既存のレコードの関係によって異なります。

このユニットでは、INSERT を使用して新しい行を追加し、データを上書き操作に置き換え、アップサート シナリオで MERGE を使用して変更を同期する方法について説明します。

INSERT INTO を使用してデータを追加する

最も簡単な読み込み操作では、現在のデータを変更せずに、既存のテーブルに新しい行を追加します。 INSERT INTO は、毎日のトランザクション ログ、新しいイベント ストリーム、増分データ抽出など、既存のレコードと重複しないデータの新しいバッチを読み込む場合に使用します。

INSERT INTO は、SQL と PySpark の両方で機能します。 次の例は、売上テーブルに行を追加する方法を示しています。

-- Append new records using VALUES
INSERT INTO sales.transactions (transaction_id, amount, transaction_date)
VALUES 
    ('TXN001', 150.00, '2026-01-15'),
    ('TXN002', 275.50, '2026-01-15');

-- Append from a staging table or query
INSERT INTO sales.transactions
SELECT * FROM staging.daily_transactions
WHERE transaction_date = current_date();

PySpark では、 append モードで DataFrame を書き込んでデータを追加します。

# Append DataFrame to existing table
df_new_transactions.write.mode("append").saveAsTable("sales.transactions")

# Or use insertInto for existing tables
df_new_transactions.write.insertInto("sales.transactions")

データを INSERT OVERWRITE に置き換える

既存のデータを追加するのではなく、置き換える必要がある場合があります。 INSERT OVERWRITE では、新しい行を挿入する前に、ターゲット テーブルまたはパーティションが切り捨てられます。 この方法は、失敗したバッチの再処理、参照テーブルの更新、集計の再構築に適しています。

-- Replace entire table contents
INSERT OVERWRITE sales.daily_summary
SELECT 
    region,
    SUM(amount) as total_sales,
    COUNT(*) as transaction_count
FROM sales.transactions
WHERE transaction_date = current_date()
GROUP BY region;

パーティション テーブルの場合は、特定のパーティションを上書きし、他のパーティションはそのまま残すことができます。

-- Replace only January 2026 partition
INSERT OVERWRITE sales.monthly_report
PARTITION (report_month = '2026-01')
SELECT region, product_category, SUM(amount) as revenue
FROM sales.transactions
WHERE transaction_date BETWEEN '2026-01-01' AND '2026-01-31'
GROUP BY region, product_category;

Delta Lake では、 REPLACE WHERE 句による選択的上書きもサポートされています。これにより、挿入する前に条件に一致する行が削除されます。

-- Replace transactions for a specific date range
INSERT INTO sales.transactions
REPLACE WHERE transaction_date >= '2026-01-01' AND transaction_date <= '2026-01-07'
SELECT * FROM staging.corrected_transactions;

既定では、クエリから返されるすべての行は、述語 REPLACE WHERE と一致する必要があります。 いずれかの行が述語条件を満たさない場合、ステートメントは失敗し、何も書き込みません。

PySpark では、テーブルの完全な置換に overwrite モードを使用し、条件に一致する行のみを置き換える replaceWhere オプションを追加します。

# Overwrite entire table
df_refreshed.write.mode("overwrite").saveAsTable("sales.daily_summary")

# Replace only the January 2026 rows
(df_january.write
    .mode("overwrite")
    .option("replaceWhere", "report_month = '2026-01'")
    .saveAsTable("sales.monthly_report"))

REPLACE USING を使用してデータを動的に置換する

REPLACE USING は、動的データの上書きに推奨される方法です。 Databricks SQL ウェアハウスやサーバーレス コンピューティングなど、すべてのコンピューティングの種類で機能し、Spark セッションの構成は必要ありません。 partitionOverwriteModeとは異なり、REPLACE USINGでは、パーティション テーブル、パーティション分割されていないテーブル、および液体クラスタリングを使用したテーブルがサポートされます。

-- Replace only the partitions touched by the incoming data
INSERT INTO sales.monthly_report
  REPLACE USING (report_month)
  SELECT region, product_category, SUM(amount) AS revenue, report_month
  FROM staging.corrected_transactions
  GROUP BY region, product_category, report_month;

REPLACE USING句は、sales.monthly_reportの受信行と一致するreport_month内の行をアトミックに削除し、新しい行を挿入します。 指定した列と一致しない行は変更されません。 Databricks Runtime 16.3 –17.1 では、 USING 列はテーブルのパーティション列の完全なセットである必要があります (レガシ動作)。 Databricks Runtime 17.2 以降では、パーティション分割されていないテーブルや液体クラスタリングを使用するテーブルなど、任意のテーブルの種類の列がサポートされています。 SQL REPLACE USING句には Databricks Runtime 16.3 以上が必要であり、Pythonと Scala replaceUsing オプションには Databricks Runtime 18.2 以降が必要です。 REPLACE USINGでは通常の等価性が使用されるため、NULL列にUSING値を持つ行は一致せず、ターゲット テーブルから削除されません。

カスタム条件に基づいて行を置き換えたり、NULL 値を等しい値として扱ったりするなど、より柔軟な照合が必要な場合は、代わりに REPLACE ON を使用します。

-- Replace rows matching a user-defined condition across source and target
INSERT INTO sales.transactions AS target
  REPLACE ON (target.transaction_id = source.transaction_id)
  (SELECT * FROM staging.corrected_transactions) AS source;

REPLACE ONには、SQL の場合は Databricks Runtime 17.1 以降、Pythonおよび Scala replaceOn オプションには Databricks Runtime 18.2 以降が必要です。 ソース テーブル参照なしで条件ベースの置換が必要な場合は、 REPLACE WHERE (前述) を使用します。

アップサートの操作のためにデータをマージする

実際のデータ パイプラインでは、多くの場合、新しいレコードを挿入しながら既存のレコードを更新する必要があります。 MERGE ステートメントは、ソース行をターゲット行と照合し、一致が存在するかどうかに基づいて異なるアクションを適用することで、このパターン (一般に "upsert" と呼ばれます) を処理します。

MERGE は、緩やかに変化するディメンション、ソース システムからのデータの同期、変更データ キャプチャ (CDC) フィードの処理に不可欠です。

MERGE INTO customers AS target
USING customer_updates AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET 
        target.email = source.email,
        target.phone = source.phone,
        target.last_updated = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, phone, created_date, last_updated)
    VALUES (source.customer_id, source.name, source.email, source.phone, 
            current_timestamp(), current_timestamp());

このクエリは、まったく新しい顧客レコードを挿入しながら、既存の顧客の電子メールと電話を更新します。 ON 句は、一致するキー (通常は主キーまたはビジネス識別子) を定義します。

条件付きロジックを使用してマージする

より複雑なシナリオを処理するための条件を WHEN 句に追加します。 たとえば、実際に変更されたレコードのみを更新できます。

MERGE INTO products AS target
USING product_feed AS source
ON target.sku = source.sku
WHEN MATCHED AND source.price <> target.price THEN
    UPDATE SET target.price = source.price, target.updated_at = current_timestamp()
WHEN MATCHED AND source.discontinued = true THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT *;

INSERT *構文では、ソースのすべての列がターゲット スキーマと一致したときに挿入されます。

PySpark との結合

プログラムによる制御には、Delta Lake Python API を使用します。

from delta.tables import DeltaTable

target_table = DeltaTable.forName(spark, "customers")

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdate(set={
    "email": "source.email",
    "phone": "source.phone",
    "last_updated": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "customer_id": "source.customer_id",
    "name": "source.name",
    "email": "source.email",
    "phone": "source.phone",
    "created_date": "current_timestamp()",
    "last_updated": "current_timestamp()"
}).execute()

ソース スキーマとターゲット スキーマが一致する場合は、便利なメソッドを使用します。

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdateAll(
).whenNotMatchedInsertAll(
).execute()

適切な読み込み戦略を選択する

読み込み操作の選択は、ソース データとターゲット データの関係によって異なります。

これらの読み込みパターンを理解すると、新しいレコードと変更されたレコードの両方を効率的に処理しながら、データの整合性を維持する信頼性の高いデータ パイプラインを構築する準備が整います。