Lakeflow pipelines のベストプラクティス

新しいパイプラインを開始する場合でも、既存のパイプラインを改善する場合でも、パイプラインの設計、構築、運用時にはこれらの推奨パターンを適用してください。

次のページでは、具体的な設計上の決定事項についてさらに詳しく解説します。

トピック	説明
LakeFlow Pipelines におけるディメンショナルモデリング	ゴールドレイヤーのデータをスタースキーマのファクトおよびディメンションとしてモデル化し、ディメンションモデリングをパイプラインのデータセット型にマッピングします。
LakeFlow Pipelines における処理保証	べき等性と、"default"で「exactly-once」処理が適用される箇所、およびべき等書き込みを追加する必要がある境界条件について理解します。
LakeFlow Pipelines全体でデータセットを整理する	1つのパイプラインにどれだけのデータセットを割り当てるか、またいつ作業を別々のパイプラインに分割するかを決めてください。
LakeFlow Pipelines の本番運用準備	パイプラインを無人で実行する前に、データ品質、信頼性、可観測性、デプロイメント、コスト、ガバナンスに関するチェックリストを確認してください。

トピック	説明
LakeFlow Pipelines におけるディメンショナルモデリング	ゴールドレイヤーのデータをスタースキーマのファクトおよびディメンションとしてモデル化し、ディメンションモデリングをパイプラインのデータセット型にマッピングします。
LakeFlow Pipelines における処理保証	べき等性と、"default"で「exactly-once」処理が適用される箇所、およびべき等書き込みを追加する必要がある境界条件について理解します。
LakeFlow Pipelines全体でデータセットを整理する	1つのパイプラインにどれだけのデータセットを割り当てるか、またいつ作業を別々のパイプラインに分割するかを決めてください。
LakeFlow Pipelines の本番運用準備	パイプラインを無人で実行する前に、データ品質、信頼性、可観測性、デプロイメント、コスト、ガバナンスに関するチェックリストを確認してください。

適切なデータセットタイプを選択してください

パイプラインはストリーミングテーブル、マテリアライズドビュー、一時ビューの3種類のデータセットを提供します。パイプラインの各層に適切なタイプを選ぶことで、不要なコンピュートコストを避け、コードを読みやすくします。

ストリーミングテーブル は、データ取り込みおよび低レイテンシーのストリーミング変換に適しています。各入力行は一度だけ読み取られて処理されるため、追加専用のワークロード、大容量データ、クラウドストレージやメッセージバスからのイベント駆動型処理に最適です。

マテリアライズドビュー は複雑な変換や解析クエリに適した選択肢です。その結果は事前に計算され、インクリメンタルリフレッシュで常に更新されるため、それらに対するクエリーは高速です。マテリアライズドビューではデータを直接変更することはできません。クエリー定義は出力を制御します。

一時ビュー は、データをストレージにマテリアライズすることなく変換ロジックを整理する、パイプラインスコープのビューです。独自のテーブルを必要としない中間ステップに使用します。

次の表は、各タイプをいつ使用するかをまとめたものです。

ユースケース	推奨タイプ	理由:
クラウドストレージやメッセージバスからの取り込み	ストリーミングテーブル	各レコードを一度だけ処理します。大容量および追加専用のワークロードを処理します。
CDCストリーム（挿入、更新、削除）	ストリーミングテーブル	順序付けられ、重複排除された CDC 取り込みのための AUTO CDC ... INTO のターゲットとして使用されます。
複雑な集計と結合	マテリアライズドビュー	増分更新されます。更新のたびに完全な再計算を行う必要はありません。
ダッシュボードのクエリー高速化	マテリアライズドビュー	事前計算された結果を使用することで、生のテーブルに対してクエリーを実行するよりも高速になります。
中間変換（下流のリーダーなし）	一時的ビュー	ストレージコストを発生させずにパイプラインロジックを整理できます。

ユースケース	推奨タイプ	理由:
クラウドストレージやメッセージバスからの取り込み	ストリーミングテーブル	各レコードを一度だけ処理します。大容量および追加専用のワークロードを処理します。
CDCストリーム（挿入、更新、削除）	ストリーミングテーブル	順序付けられ、重複排除された CDC 取り込みのための AUTO CDC ... INTO のターゲットとして使用されます。
複雑な集計と結合	マテリアライズドビュー	増分更新されます。更新のたびに完全な再計算を行う必要はありません。
ダッシュボードのクエリー高速化	マテリアライズドビュー	事前計算された結果を使用することで、生のテーブルに対してクエリーを実行するよりも高速になります。
中間変換（下流のリーダーなし）	一時的ビュー	ストレージコストを発生させずにパイプラインロジックを整理できます。

詳細については、「ストリーミングテーブル」「マテリアル化されたビュー」「LakeFlow Pipelinesとは何か?」をご覧ください。

命令型 MERGE の代わりに宣言型 CDC を使用する

命令型SQL MERGE 文を用いたチェンジデータキャプチャ(CDC)を実装するには、イベントの順序付け、重複除去、部分的な更新、スキーマ進化を正しく扱うために大量のカスタムコードが必要です。これらの懸念事項はそれぞれ個別に解決する必要があり、結果として得られるコードは保守やテストが困難になる。

パイプラインは、AUTO CDC ... INTO ステートメント (SQL) と create_auto_cdc_flow() 関数 (Python) を提供し、これらによって順序付け、重複排除、順不同イベント、スキーマ進化を宣言的に処理します。変更フィードの形状とターゲットテーブルを記述し、パイプラインが残りを処理します。AUTO CDC SCDタイプ1(上書き)とSCDタイプ2(履歴保存)の両方をサポートしています。

詳細については、チェンジデータキャプチャとスナップショットおよび「AUTO CDC APIs: パイプラインによるチェンジデータキャプチャの簡素化」を参照してください。

エクスペクテーションを使用してデータ品質を適用する

期待値は、データセットを通過するすべての行に適用されるtrue/falseのSQL式です。行が条件に失敗した場合、パイプラインは設定した違反ポリシーに従って応答します。期待値はポリシーに関わらずパイプラインイベントログにメトリクスを送るため、時間経過によるデータ品質の傾向を追跡できます。

違反ポリシーを選択

3 つの違反ポリシーが利用可能です。不正なデータに対する許容範囲に一致するものを選択してください：

• warn (default): 無効なレコードはターゲットテーブルに書き込まれ、メトリクスにフラグが立てられます。すべてのデータをキャプチャする必要があるものの、品質問題の可視性を確保したい場合にこのポリシーを使用します。
• drop : 無効なレコードは書き込み前に破棄されます。不正な行が発生する可能性があり、それをダウンストリームに反映させたくない場合に使用します。
• fail : パイプラインの更新は、最初の無効なレコードで停止します。不正なレコードが深刻なアップストリームの問題を示すような、重要なデータに対してこれを使用してください。

以下の例は、ストリーミングテーブルに適用される各ポリシーを示しています。

SQL

SQL

-- Warn: write invalid records but track them in metrics
CREATE OR REFRESH STREAMING TABLE orders_raw (
  CONSTRAINT valid_order_id EXPECT (order_id IS NOT NULL)
) AS SELECT * FROM STREAM read_files("/volumes/raw/orders", format => "json");

-- Drop: discard invalid records before writing
CREATE OR REFRESH STREAMING TABLE orders_clean (
  CONSTRAINT non_negative_amount EXPECT (amount >= 0) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(orders_raw);

-- Fail: stop the pipeline on any invalid record
CREATE OR REFRESH STREAMING TABLE orders_critical (
  CONSTRAINT required_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM(orders_clean);

Python

Python

from pyspark import pipelines as dp

# Warn: write invalid records but track them in metrics
@dp.table
@dp.expect("valid_order_id", "order_id IS NOT NULL")
def orders_raw():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/volumes/raw/orders")

# Drop: discard invalid records before writing
@dp.table
@dp.expect_or_drop("non_negative_amount", "amount >= 0")
def orders_clean():
    return spark.readStream.table("orders_raw")

# Fail: stop the pipeline on any invalid record
@dp.table
@dp.expect_or_fail("required_customer_id", "customer_id IS NOT NULL")
def orders_critical():
    return spark.readStream.table("orders_clean")

隔離無効記録

削除されたレコードを黙って破棄するのではなく、調査のために保持したい場合は、隔離パターンを使用します。2 つのフローを使用して、検証に失敗した行を別のストリーミングテーブルにルーティングします。1 つはメインテーブルから無効な行を削除するフロー、もう 1 つは無効な行のみを隔離テーブルに書き込むフローです。これにより、クリーンなデータセットを汚染することなく、不正なデータを調査、修正、再処理できます。

クアランティンパターンの詳細な例については、エクスペクテーションの推奨事項と高度なパターンをご覧ください。

エクスペクテーションの詳細については、「パイプラインのエクスペクテーションを使用してデータ品質を管理する」を参照してください。

パイプラインのパラメータ化

パイプラインにはdefaultのカタログとスキーマの設定があるため、同じカタログとスキーマ内で読み書きを行うコードは、パラメーターなしで環境間をまたいで動作します。ただし、パイプラインが2つ目のカタログやスキーマを参照する必要がある場合 (例: 開発環境と本番環境で異なる共有ソースカタログから読み取る場合)、それらの名前をソースコードに直接ハードコーディングすることは避けてください。代わりに、それらをパイプライン構成パラメーター (パイプライン設定で設定されるキーと値のペア) として定義し、コード内で参照してください。これにより、パラメーター値を入れ替えることで、単一のコードベースを環境間で正しく実行できるようになります。

SQL

SQL

CREATE OR REFRESH MATERIALIZED VIEW transaction_summary AS
SELECT account_id, COUNT(txn_id) AS txn_count, SUM(amount) AS total_amount
FROM ${source_catalog}.sales.transactions
GROUP BY account_id;

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import count, sum

@dp.materialized_view
def transaction_summary():
    source_catalog = spark.conf.get("source_catalog")
    return spark.read.table(f"{source_catalog}.sales.transactions") \
        .groupBy("account_id") \
        .agg(
            count("txn_id").alias("txn_count"),
            sum("amount").alias("total_amount")
        )

詳細については、「 パイプラインでパラメータを使う」をご覧ください。

Triggerと連続パイプラインモードのどちらかを選べます

Trigger mode は、利用可能なすべてのデータを処理した後に停止します。これは、大多数のパイプライン（スケジュール（時間単位、日単位、またはオンデマンド）に基づいて実行され、1分未満のデータの鮮度を必要としないパイプライン）にとって適切な選択肢です。

連続モード では、クラスターが実行され続け、新しいデータが到着するたびに処理されます。これは、ユースケースで数秒から数分単位のレイテンシが求められる場合にのみ適しています。連続モードでは常時稼働のクラスターが必要となるため、Triggerモードよりも大幅にコストが高くなります。

リアルタイムモード は連続モードを基盤とし、不正検出やリアルタイムのパーソナライズなどの運用ワークロードに対してサブ秒、ミリ秒単位の遅延を実現します。追加の設定とコンピュート計画が必要です。「 Lakeflow Pipelinesのリアルタイムモード使用」を参照してください。

詳細については、Trigger パイプライン モードと継続的パイプライン モードおよびパイプラインの構成を参照してください。

データLayoutにはリキッドクラスタリングを使います

リキッドクラスタリングは、Delta テーブルのデータLayoutを最適化するために、静的パーティショニングと ZORDER に代わるものです。静的パーティショニングでは、パーティション列を選択して事前にデータを再編成する必要があり、値が不均等に分散しているとデータスキューが発生する可能性があります。リキッドクラスタリングは自己調整型で、スキュー耐性があり、インクリメンタル（増分）です。実行のたびに、再編成が必要なデータのみを書き換えます。

クエリーパターンが進化しても、テーブル全体を書き換えることなく、いつでもクラスタリング列を変更できます。

Databricks では、自動リキッドクラスタリングを推奨しています。これにより、Databricks はクエリーのワークロードに基づいて最適なクラスタリング列を自動的に選択および維持します。CLUSTER BY AUTO で有効にする：

SQL

SQL

CREATE OR REFRESH STREAMING TABLE events
CLUSTER BY AUTO
AS SELECT * FROM STREAM read_files("/volumes/raw/events", format => "parquet");

Python

Python

from pyspark import pipelines as dp

@dp.table(cluster_by_auto=True)
def events():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "parquet") \
        .load("/volumes/raw/events")

クラスタリング列を自分で選ぶには、明示的に指定してください:

SQL

SQL

CREATE OR REFRESH STREAMING TABLE events
CLUSTER BY (event_date, region)
AS SELECT * FROM STREAM read_files("/volumes/raw/events", format => "parquet");

Python

Python

from pyspark import pipelines as dp

@dp.table(cluster_by=["event_date", "region"])
def events():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "parquet") \
        .load("/volumes/raw/events")

詳細については、「ストリーミングテーブル」および「テーブルにリキッドクラスタリングを使用する」を参照してください。

CI/CDおよび宣言的自動化バンドルでパイプラインを管理

パイプラインのソースコードをバージョン管理し、宣言 的自動化バンドル を使って環境間の展開を管理しましょう。

詳細については、「ソース管理されたパイプラインの作成」、「パイプラインのバンドルプロジェクトへの変換」、および「パイプラインでのパラメーターの使用」を参照してください。

パイプラインコードをバージョン管理に格納する

すべてのパイプラインソースファイル(PythonやSQL)は、バンドル設定と一緒にGitリポジトリに保存してください。プロジェクト全体のバージョン管理は、変更履歴を完全に把握でき、コラボレーションを容易にし、開発環境で変更を検証してから本番運用に昇格させることができます。

Databricksはこのワークフロー管理に Declarative Automation Bundles を推奨しています。バンドルはソースコードとともにYAMLでパイプライン構成を定義し、 databricks bundle CLIはターミナルやCI/CDシステムからパイプラインの検証、デプロイ、実行を可能にします。

環境分離にはバンドルターゲットを使用する

バンドルは複数の ターゲット (例: dev、 staging、 prod)を有効にし、それぞれカタログ名、クラスターポリシー、通知アドレス、その他の設定に対する独自のオーバーライドセットを持ちます。バンドルターゲットとパイプラインパラメータを組み合わせて、デプロイ時に正しい環境固有の値を注入し、ソースコードを環境定数から解放します。

一般的なワークフローは以下のとおりです。

1. 開発者はフィーチャーBranchで作業し、devカタログ内の個人用開発パイプラインにデプロイします。
2. メインBranchへのMerge時に、CIシステムが databricks bundle validate と databricks bundle deploy --target staging を実行し、パイプラインを検証してステージング環境にデプロイします。
3. テストに合格した後、CI システムは databricks bundle deploy --target prod を使用して本番運用にデプロイします。

ストリーミングのベストプラクティス

これらのパターンを使用して、状態を管理し、遅延データを制御し、ストリーミングパイプラインの信頼性を維持します。

詳細については、「 ウォーターマークによるステートフル処理の最適化」「 ストリーミングチェックポイント障害からのパイプラインの回復」「 パイプラインによるヒストリカルデータのバックフィル」をご覧ください。

ステート操作にウォーターマークを使用する

ウォーターマークは、ウィンドウ集計や重複排除などのステートフル ストリーミング操作中にパイプラインがメモリ内に保持する状態を制限します。ウォーターマークがない場合、パイプラインがすべての可能なキーに対してデータを蓄積するにつれて状態が際限なく増加し、最終的に長時間実行されるパイプラインでメモリ不足エラーが発生します。

ウォーターマークは、Timestamp列と遅延データの許容thresholdを指定します。thresholdを超過した後に到着したレコードはドロップされます。遅延データに対する許容範囲と、その状態を保持するためのメモリコストのバランスが取れたthresholdを選択してください。

次の例では、3 分のウォーターマークを使用して 1 分のタンブリングウィンドウ集計をコンピュートします：

SQL

SQL

CREATE OR REFRESH STREAMING TABLE event_counts AS
SELECT window(event_time, '1 minute') AS time_window, region, COUNT(*) AS cnt
FROM STREAM(events_raw)
  WATERMARK event_time DELAY OF INTERVAL 3 MINUTES
GROUP BY time_window, region;

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import window

@dp.table
def event_counts():
    return (
        spark.readStream.table("events_raw")
            .withWatermark("event_time", "3 minutes")
            .groupBy(window("event_time", "1 minute"), "region")
            .count()
    )

注記

集約が各更新ごとに完全に再計算されるのではなく、段階的に処理されるようにするためには、ウォーターマークを定義する必要があります。

ストリーミング状態と完全更新の理解

ストリーミング状態はインクリメンタルです。パイプラインは毎回最初から再計算するのではなく、更新全体にわたって状態を構築および維持します。これがステートフルなストリーミングを効率的にする理由ですが、同時に、ステートフルなクエリーのロジックを変更した場合（例：ウォーターマークの threshold を変更したり、集計列を変更したりする場合）、既存の状態が新しいロジックと互換性を持たなくなることを意味します。この場合、新しいロジックですべてのヒストリカルデータを再処理し、状態を最初から再構築するために、フル更新を実行する必要があります。

ソースがヒストリカルデータを保持していない場合、フル更新によってデータが失われる可能性もあります。例えば、保持期間が短い Kafka ソースでは、更新時に直近数分間のデータしか利用できない場合があり、その結果、以前よりもはるかに少ないデータしか含まれないテーブルになる可能性があります。ステートフルなクエリーロジックの変更は慎重に計画してください。特に、フル更新のコストが高い場合や、ソースのデータ保持期間が短い大容量のストリームでは注意が必要です。メダリオンアーキテクチャを使用すると、最小限の変換でブロンズテーブルを作成でき、シルバーテーブルやゴールドテーブルが完全な履歴を使用してブロンズテーブルから再計算できるようになるため、有用です。

ストリーム-ストリームJOIN

ストリーム-ストリーム結合では、結合の 両側 にウォーターマークと時間制限のある結合条件が必要です。結合条件内の時間間隔は、それ以上の一致が不可能になるタイミングをストリーミングエンジンに通知し、一致しなくなった状態を削除できるようにします。ウォーターマークまたは時間制限条件のいずれかを省略すると、状態は無制限に増加します。

次の例では、広告インプレッションイベントとクリックイベントを結合します。その際、インプレッションから3分以内にクリックが発生することを条件としています。

SQL

SQL

CREATE OR REFRESH STREAMING TABLE impression_clicks AS
SELECT imp.ad_id, imp.impression_time, clk.click_time
FROM STREAM(ad_impressions)
    WATERMARK impression_time DELAY OF INTERVAL 3 MINUTES AS imp
JOIN STREAM(user_clicks)
    WATERMARK click_time DELAY OF INTERVAL 3 MINUTES AS clk
ON imp.ad_id = clk.ad_id
  AND clk.click_time BETWEEN imp.impression_time
    AND imp.impression_time + INTERVAL 3 MINUTES;

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import expr

dp.create_streaming_table("impression_clicks")

@dp.append_flow(target="impression_clicks")
def join_impressions_and_clicks():
    impressions = spark.readStream.table("ad_impressions") \
        .withWatermark("impression_time", "3 minutes")
    clicks = spark.readStream.table("user_clicks") \
        .withWatermark("click_time", "3 minutes")
    return impressions.alias("imp").join(
        clicks.alias("clk"),
        expr("""
            imp.ad_id = clk.ad_id AND
            clk.click_time BETWEEN imp.impression_time AND imp.impression_time + INTERVAL 3 MINUTES
        """),
        "leftOuter"
    )

ストリームと静的テーブルを結合する場合 (スナップショット結合)、静的テーブルのスナップショットは各マイクロバッチの開始時に更新されます。つまり、遅れて到着したディメンションレコードは、すでに処理済みのファクトには遡及的に適用されません。遡及的な適用が必要な場合は、マテリアライズドビューを使用するか、パイプラインを再構築してください。

パイプライン性能の最適化

これらの手法を適用して、コンピュートコストを削減し、パイプラインの更新を高速化します。

詳細については、「マテリアライズドビュー」および「ウォーターマークを使用したステートフル処理の最適化」を参照してください。

小さなファイルを避ける

低ボリュームのソースに対してパイプラインを頻繁にトリガーすると、クラウドストレージに多数の小さなファイルが書き込まれます。小さなファイルは、ファイルごとに個別のメタデータ検索とI/Oラウンドトリップが必要となり、クラウドストレージの APIs が大規模なリスト操作をスロットルするため、読み取りパフォーマンスを低下させます。これを回避するには、データボリュームに合わせたTrigger間隔を選択してください。継続的に実行するのではなく、更新の間に十分な量のデータが蓄積されるようなスケジュールでTriggerされたパイプラインを実行します。

データスキューの処理

データスキューは、結合キーまたはgroupByキーの値がパーティション間で不均等に分散され、少数のタスクがデータの大部分を処理することになった場合に発生します。これによりホットスポットが作成され、エンドツーエンドの更新時間が増加します。保存されたテーブルのスキューに対処するには、リキッドクラスタリングを使用します。実行中の計算中に発生するスキューについては、2段階でグループ化および集計を行う前に、ランダムなバケットサフィックスを付加して、偏りの大きいキーをソルティングします。

詳細については、 「データ**Layout**に液体クラスタリングを使用する」を参照してください。

マテリアライズドビューにはインクリメンタル更新を活用してください

大規模な集約にマテリアライズドビューを使うと、パイプラインは段階的に更新しようとし、前回の更新以降の上流の変更のみを処理し、結果セット全体を再計算しません。インクリメンタル更新は、各パイプライン Trigger でクエリーを一からやり直すよりもはるかに安価です。マテリアライズドビューを段階的に更新できる可能性を最大化するために、シンプルで決定性的な集計クエリーを書き、非決定性関数のような増分処理を妨げる構成要素は避けてください。

マテリアライズドビューの増分更新を参照してください。

結合の最適化

一方が小さな次元テーブルであるジョインの場合、シャッフルジョインを行う代わりにSparkに小さなテーブルを全エグゼキューターにブロードキャストするよう指示するブロードキャストヒントを追加してください:

SQL

SQL

CREATE OR REFRESH MATERIALIZED VIEW enriched_orders AS
SELECT o.*, /*+ BROADCAST(p) */ p.product_name, p.category
FROM orders o
JOIN products p ON o.product_id = p.product_id;

Python

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import broadcast

@dp.materialized_view
def enriched_orders():
    orders = spark.read.table("orders")
    products = spark.read.table("products")
    return orders.join(broadcast(products), "product_id")

時系列近接結合（例：時間範囲内で最も近いイベントを検索する場合）では、範囲結合条件を使用し、ストリームを結合する場合は両側にウォーターマークがあることを確認するか、結合前にイベントを時間バケットに事前ビン分割することを検討してください。

パイプラインの監視

パイプラインイベントLogsは、パイプラインにおける主要なオブザーバビリティのプリミティブです。各パイプラインのランは、実行進行状況、データ品質の期待値、データリネージ、エラーの詳細を含む構造化レコードをイベントログに書き込みます。イベントログは、直接クエリー可能なDeltaテーブルです。

基盤となるストレージパスを知らずにイベント log をクエリするには、共有クラスターまたは SQL Warehouse 上で event_log() テーブル値関数を使用します。

SQL

SELECT * FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC
LIMIT 100;

イベント Logs でエクスペクテーションメトリクスをクエリして、データ品質ダッシュボードを構築します。details 列には、各制約の成功/失敗カウントを含むネストされたJSON構造が含まれており、これを使用して経時的な品質傾向を追跡し、リグレッション（回帰）に関するアラートを設定できます。

イベント駆動型アラートでは、パイプラインが失敗したりデータ品質のthresholdが突破された際に、イベントフックを使ってカスタムウェブフックや通知サービス(SlackやPagerDutyなど)をTriggerします。イベントフックとは、パイプラインイベントに応じて実行されるPython関数のことです。

詳細については、「 パイプラインの監視」「 パイプラインイベントLogs」「 イベントフックを用いたパイプラインのカスタムモニタリングの定義」をご覧ください。

Use Serverless コンピュートを使用する

Databricksでは、新しいパイプラインに対してServerless コンピュートの使用を推奨しています。Serverlessでは手動のクラスター構成は不要であり、Databricksがインフラストラクチャを自動的に管理します。Serverless パイプラインは、ワークロードの需要に応じて水平方向（エグゼキューターの増加）と垂直方向（エグゼキューターサイズの拡大）の両方にスケーリングできる強化オートスケールを使用します。Serverless パイプラインは常に Unity Catalog を使用するため、ガバナンスとリネージ追跡がdefaultで組み込まれています。

マテリアライズドビューの増分更新にもServerlessが必要です。クラシックコンピュートでは、マテリアライズドビューは常に完全に再計算されるため、更新コストが増加します。ワークロードにおいて増分更新が重要な場合は、Serverlessコンピュートを使用してください。

Serverless とクラシック コンピュートの比較については、「パイプラインにおける Serverless とクラシック コンピュートの比較」を参照してください。Serverlessの詳細については、Serverless パイプラインの設定を参照してください。

メダリオンアーキテクチャを使用してパイプラインを整理する

メダリオンアーキテクチャは、データを3つの論理レイヤー（ブロンズ、シルバー、ゴールド）に整理し、それぞれに明確な目的を持たせています。パイプラインのデータセットタイプを適切なレイヤーにマッピングすることで、各レイヤーの責任が明確になり、パイプラインの保守が容易になります。

• ブロンズ : ストリーミングテーブルを使用して、クラウドストレージ、メッセージバス、またはCDCソースから生データを取り込みます。ブロンズテーブルは、最小限の変換で生ソースデータを保持するため、要件が変更された場合にシルバーレイヤーやゴールドレイヤーがブロンズレイヤーのソースから再処理を行うことが可能です。
• シルバー : インクリメンタルな行レベルの変換（フィルタリング、クリーニング、解析）には、ストリーミングテーブルを使用します。シルバーレイヤーのロジックにディメンションテーブルに対するエンリッチメント結合や、増分更新のメリットがある複雑な集計が含まれる場合は、マテリアライズドビューを使用します。
• ゴールド : マテリアライズドビューを使用して、ダッシュボード、レポートツール、およびダウンストリームのコンシューマーに提供される集計、メトリクス、およびサマリーを事前計算します。

可能な限り、取り込み（ブロンズ）と変換（シルバーおよびゴールド）を別々のパイプラインに分離してください。レイヤーを分離することで、各レイヤーのスケジュール、監視、トラブルシューティングを独立して行えるようになります。また、変換パイプラインで障害が発生しても、新しいデータがブロンズに格納されることは妨げられません。

詳細については、ストリーミングテーブルおよびマテリアライズドビューを参照してください。