パイプラインの更新を実行する
パイプラインの更新は、クラスターを起動し、ソースコードを検証し、パイプラインで定義されているテーブルとビューを更新します。更新は手動で、スケジュールで、またはプログラムでトリガーできます。
パイプライン更新とは何ですか?
パイプラインを作成し、実行する準備ができたら、 更新 を開始します。パイプラインの更新では次の処理が行われます。
- 正しい構成でクラスターを開始します。
- 定義されているすべてのテーブルとビューを検出し、無効な列名、欠落している依存関係、構文エラーなどの分析エラーがないかチェックします。
- 使用可能な最新のデータでテーブルとビューを作成または更新します。
検証 を使用すると、テーブルが作成または更新されるのを待たずに、パイプラインのソースコードに問題がないか確認できます。この機能は、パイプラインの開発またはテスト時に、テーブル名や列名の誤りなど、パイプラインのエラーをすばやく見つけて修正できるため便利です。
パイプラインの更新はどのようにトリガーされますか?
パイプラインの更新を開始するには、次のいずれかのオプションを使用します。
更新トリガー | 詳細 |
|---|---|
手動 | Lakeflow Pipelines Editor またはパイプライン リストからパイプラインの更新を手動でトリガーできます。 「パイプラインの更新を手動でトリガーする」を参照してください。 |
スケジュール | ジョブを使用してパイプラインの更新をスケジュールできます。ジョブのパイプライン タスクを参照してください。 |
プログラム | サードパーティのツール、 APIs 、CLI を使用して、プログラムで更新をトリガーできます。 「ワークフローのパイプライン」および「パイプラインREST API実行」を参照してください。 |
パイプラインの更新を手動でトリガーする
パイプラインの更新を手動でトリガーするには、次のいずれかのオプションを使用します。
- Lakeflow Pipelines Editor から完全なパイプライン、またはパイプラインのサブセット (単一のソース ファイル、または単一のテーブル) を実行します。 詳細については、 「実行パイプライン コード」を参照してください。
- ジョブとパイプラインの リストから完全なパイプラインを実行します。 クリック
リスト内のパイプラインと同じ行にあります。 - パイプラインモニタリングページから、
ボタン。
手動でトリガーされたパイプライン更新のデフォルトの動作は、パイプラインで定義されているすべてのデータセットを更新することです。
パイプライン更新セマンティクス
次の表は、マテリアライズドビューとストリーミング テーブルの確実更新、完全更新、およびリセット チェックポイントの動作を説明しています。
アップデートのタイプ | マテリアライズドビュー | ストリーミングテーブル |
|---|---|---|
更新（デフォルト） | 定義クエリの現在の結果を反映するように、結果を更新します。Databricks はコストを評価し、より効率的である場合に増分更新を実行します。マテリアライズドビューの増分更新を参照してください。 | 新しいレコードをストリーミングテーブルおよびフローに定義されたロジックを通じて処理します。 |
フルリフレッシュ | 定義クエリの現在の結果を反映するために、結果を再計算します。 | ストリーミングテーブルからデータをクリアし、フローからチェックポイントをクリアし、データソースからすべてのレコードを再処理します。ストリーミングテーブルのフル更新を参照してください。 |
ストリーミングフローのチェックポイントをリセットする | マテリアライズドビューには適用されません。 | フローからチェックポイントをクリアしますが、ストリーミングテーブルからデータはクリアしません。その後、データソースからすべてのレコードを再処理します。 |
デフォルトでは、パイプライン内のすべてのマテリアライズドビューとストリーミングテーブルは、更新のたびに更新されます。 オプションで、次の機能を使用して更新からテーブルを省略できます。
- 更新するテーブルを選択 : 更新を実行する前に、この UI を使用してマテリアライズドビューとストリーミング テーブルを追加または削除します。 「選択したテーブルのパイプラインの更新を開始する」を参照してください。
- 失敗したテーブルの更新 : ダウンストリームの依存関係を含む、失敗したマテリアライズドビューとストリーミング テーブルの更新を開始します。 「失敗したテーブルのパイプラインの更新を開始する」を参照してください。
これら両方の機能は、デフォルトの更新セマンティクスまたは完全な更新をサポートしています。オプションで、 [更新するテーブルの選択] ダイアログを使用して、失敗したテーブルの更新を実行するときに追加のテーブルを除外できます。
ストリーミング テーブルの場合、関連付けられたストリーミング テーブルからのデータではなく、選択したフローのストリーミング チェックポイントをクリアすることを選択できます。 選択したフローのチェックポイントをクリアするには、Databricks REST API を使用して更新を開始します。選択的ストリーミング フローのチェックポイントをクリアするには、パイプラインの更新を開始するを参照してください。
フル更新を使用する場合
Databricks では、必要な場合にのみ完全更新を実行することを推奨しています。完全な更新では、常に、データセットを定義するロジックを通じて、指定されたデータ ソースからのすべてのレコードが再処理されます。 完全な更新を完了するために必要な時間とリソースは、ソース データのサイズと相関関係にあります。
マテリアライズドビューは、デフォルト更新と完全更新のどちらを使用しても同じ結果を返します。ストリーミングテーブルで完全更新を行うと、すべての状態処理とチェックポイント情報がリセットされ、入力データが利用できなくなった場合はレコードが削除される可能性があります。ストリーミングテーブルの完全更新を参照してください
Databricks 、テーブルまたはビューの目的の状態を再作成するために必要なデータが入力データ ソースに含まれている場合にのみ、完全な更新を推奨します。 入力ソース データが利用できなくなった次のシナリオと、完全更新を実行した結果について考えてみましょう。
データソース | 入力データが存在しない理由 | 完全リニューアルの成果 |
|---|---|---|
Kafka | 短期保持閾値 | Kafka ソースに存在しなくなったレコードは、ターゲット テーブルから削除されます。 |
オブジェクトストレージ内のファイル | ライフサイクルポリシー | ソース ディレクトリに存在しなくなったデータ ファイルは、ターゲット テーブルから削除されます。 |
テーブル内のレコード | コンプライアンスのため削除 | ソース テーブルに存在するレコードのみが処理されます。 |
テーブルまたはビューで完全な更新が実行されないようにするには、テーブル プロパティpipelines.reset.allowedをfalseに設定します。パイプライン テーブルのプロパティを参照してください。追加フローを使用して、完全に更新せずに既存のストリーミング テーブルにデータを追加することもできます。
選択したテーブルのパイプラインの更新を開始します
オプションで、パイプライン内の選択したテーブルのデータのみを再処理することもできます。たとえば、開発中に 1 つのテーブルのみを変更してテスト時間を短縮したい場合や、パイプラインの更新が失敗し、失敗したテーブルのみを更新したい場合などです。
Lakeflow Pipelinesエディターには、ソース ファイル、選択したテーブル、または単一のテーブルを再処理するためのオプションがあります。 詳細については、 「パイプライン コードの実行」を参照してください。
失敗したテーブルのパイプラインの更新を開始する
パイプライン グラフ内の 1 つ以上のテーブルでエラーが発生したためにパイプラインの更新が失敗した場合は、失敗したテーブルと下流の依存関係のみの更新を開始できます。
除外されたテーブルは、失敗したテーブルに依存している場合でも更新されません。
失敗したテーブルを更新するには、パイプラインモニタリングページで、 [失敗したテーブルを更新] をクリックします。
パイプライン監視ページから選択した失敗したテーブルのみを更新するには:
-
クリック
失敗したテーブルの更新 ボタンの横にあるをクリックし、 更新するテーブルの選択を クリックします。 更新するテーブルの選択 ダイアログが表示されます。 -
更新するテーブルを選択するには、各テーブルをクリックします。選択されたテーブルは強調表示され、ラベルが付けられます。テーブルを更新から削除するには、テーブルをもう一度クリックします。
-
選択範囲の更新 をクリックします。
選択範囲の更新 ボタンには、選択したテーブルの数が括弧内に表示されます。
選択したテーブルにすでに取り込まれたデータを再処理するには、
選択を更新 ボタンの横にある をクリックし、 選択の完全更新 を クリックします。
パイプラインの更新を開始して、選択的ストリーミング フローのチェックポイントをクリアします。
オプションで、すでに取り込まれたデータをクリアせずに、パイプライン内の選択したストリーミング フローのデータを再処理することもできます。
選択されていないフローは、REFRESH 更新を使用して実行されます。full_refresh_selectionまたはrefresh_selectionを指定して、他のテーブルを選択的に更新することもできます。
選択したストリーミングチェックポイントを更新する更新を起動するには、LakeFlow Pipelines REST APIで更新リクエストを使用します。
reset_checkpoint_selectionパラメーターはフロー名のリストを受け入れます。各フロー名を完全修飾catalog.schema.flow_name形式で渡す必要があります。単純な名前のみを使用すると（例: my_catalog.my_schema.my_flow の代わりに my_flow）、パイプラインの更新が失敗し、IllegalArgumentException が発生します。
- 明示的な名前でフローを定義した場合（たとえば、
create_auto_cdc_flowのflow_nameパラメーターを使用した場合）、完全修飾フロー名は<catalog>.<schema>.<flow_name>です。 - 明示的なフロー名を設定しなかった場合、デフォルトのフロー名は、
catalog.schema.table形式の完全修飾ターゲットテーブル名になります。
フロー名は、パイプラインのUIまたはパイプラインのイベントログで確認できます。
次の例では、 curlコマンドを使用してupdatesリクエストを呼び出し、パイプラインの更新を開始します。
curl -X POST \
-H "Authorization: Bearer <your-token>" \
-H "Content-Type: application/json" \
-d '{
"reset_checkpoint_selection": ["my_catalog.my_schema.my_streaming_table"]
}' \
https://<your-databricks-instance>/api/2.0/pipelines/<your-pipeline-id>/updates
以下の例では、カスタム名で定義されたフローのチェックポイントをリセットします。
curl -X POST \
-H "Authorization: Bearer <your-token>" \
-H "Content-Type: application/json" \
-d '{
"reset_checkpoint_selection": ["my_catalog.my_schema.my_custom_flow_name"]
}' \
https://<your-databricks-instance>/api/2.0/pipelines/<your-pipeline-id>/updates
テーブルの更新を待たずにパイプラインのエラーをチェックする
プレビュー
パイプラインDry run機能はパブリック プレビュー段階です。
完全な更新を実行せずにパイプラインのソースコードが有効かどうかを確認するには、 ドライ ラン を使用します。ドライ実行は、パイプラインで定義されたデータセットとフローの定義を解決しますが、データセットを具体化または公開することはありません。 dry 実行中に見つかったエラー (テーブル名や列名が正しくないなど) は UI に報告されます。
ドライ実行を開始するには、 [スタート] の横にあるパイプラインの詳細ページで [Dry 実行] を クリックします。
ドライ実行の完了後、エラーは下部のパネルのイベントトレイに表示されます。イベントトレイをクリックすると、下部のパネルに見つかった問題が表示されます。さらに、イベントログにはドライランにのみ関連するイベントが表示され、パイプライングラフにはメトリクスが表示されません。エラーが見つかった場合、詳細はイベントログで確認できます。
ドライランがパイプラインの最新の更新である場合にのみ、UIでドライラン結果を表示できます。更新履歴で選択して、結果を表示します。各更新タイプがUIで利用可能な期間については、更新結果の可用性を参照してください。
結果の可用性を更新
アップデートの結果がUIに表示されるかどうかは、アップデートの種類と以下の2つの条件によって異なります。
- 保持ウィンドウ ：完了した更新が保持される期間。パイプラインは過去60日間の更新を保持します。
- 最新の更新 ：その更新がパイプラインで起動された最新のものであるかどうか。
次の表に、各更新タイプがいつ利用可能になるかを示します：
アップデートのタイプ | UIで利用可能 | UIから削除される場合 |
|---|---|---|
定期的な更新 | 保持期間内にあるか、またはまだアクティブです（完了していません）。保持期間より前に開始された場合でも、アクティブな更新は利用可能なままです。 | 処理が完了しており、保持期間よりも古くなっています。 |
ドライラン | これはパイプラインの最新の更新です。 | それ以降の更新は、別のドライランであろうと、通常の更新であろうと、開始されます。 |
すべての場合において、更新のイベントは、その結果がUIに表示されなくなった後もイベントログに残ります。
実行動作の更新
パイプライン更新の動作は、更新をトリガーする方法によって決まります。
- 実行 を使用して パイプライン監視 UI からトリガーされる更新では、 高速起動のデバッグ重視の動作が使用されるようになりました。
- ジョブ、パイプラインAPI 、または継続的なパイプラインからトリガーされた更新では、 自動再試行および再起動動作が使用されます。
トリガーされたパイプラインの場合、 LakeFlow Pipelinesエディターまたはパイプライン モニタリング ページのドロップダウンから 別の設定で今すぐ実行を 選択することで、特定の実行の暫定動作をオーバーライドできます。
高速起動、デバッグ重視の動作
UIの 即時実行 やアドホックな更新に使用されます。これらの実行は、迅速な反復処理を最適化します。
- クラスターを再利用することで、再起動に伴うオーバーヘッドを回避します。デフォルトでは、クラスターは2時間実行されます。これは、パイプライン用のクラシック コンピュートの構成の
pipelines.clusterShutdown.delay設定で変更できます。 - パイプラインの再試行を無効にするので、エラーをすぐに検出して修正できます。
自動再試行および再起動動作
ジョブ、APIトリガーによる更新、および継続的なパイプラインに使用されます。これらの実行では、信頼性とコスト効率が優先されます。
- メモリ リークや古い資格情報など、特定の回復可能なエラーのクラスターを再開します。
- クラスターの開始の失敗など、特定のエラーが発生した場合に実行を再試行します。
- 実行が完了すると、クラスターは直ちにシャットダウンします。
実行動作はクラスターとパイプラインの実行のみを制御します。 公開テーブル用のカタログ内のストレージ場所とターゲットスキーマは、パイプライン設定の一部として構成する必要があり、実行時の動作には影響されません。