イベントログで取り込みゲートウェイの進行状況を監視する
適用対象 :
SaaSコネクタ
データベースコネクタ
イベント Logs を使用して、インジェクションゲートウェイの進行状況をリアルタイムで監視する方法を学びます。ゲートウェイはスナップショットフェーズとチェンジデータキャプチャ(CDC)フェーズの両方についてテーブルごとのメトリクスを出力するため、パイプラインの健全性を追跡し、停滞しているテーブルを特定し、構造化イベントデータからアラートとダッシュボードを作成できます。
進捗イベントでは次のことが可能になります。
- パイプラインの完了を待たずに、テーブルごとに取り込まれた行数とバイト数を追跡します。
- 大規模な初回ロードのスナップショットの進捗状況と完了予定時間を監視します。
- 各テーブルについて、ソースの commit から Unity Catalog ボリュームまでの CDC レイテンシを測定します。
- データに変更がない場合でも、パイプラインがアクティブに実行されていることを確認します。
進捗イベントの仕組み
ゲートウェイは、パイプライン内の各テーブルについて 5 分ごとに次のイベント型を出力します。最初のイベントは、パイプラインの開始から 5 分以内に表示されます。
flow_progress: スナップショットとCDCフローの両方に対して発行されます。アップサートされた行、削除された行(CDCのみ)、および前回のイベント以降の出力バイト数を報告します。CDCフローの場合は、レイテンシーメトリクスも報告します。operation_progress: スナップショット フローでのみ出力されます。スナップショット完了率とスナップショットが終了するまでの推定残り時間をレポートします。
イベントは、前のイベントから変更がないテーブルを含め、すべてのテーブルに対して発行されます。0 行のテーブルはアイドル状態です。イベントが報告されないテーブルは、パイプラインによって取得されていません。
両方のイベントタイプで METRICS ログレベルが使用されます。SQL を使用してイベントログテーブルからこれらをクエリーできます。
アクセス進行イベント
進行状況イベントはイベント ログ テーブルに保存されます。アクセスするには:
- Databricks ワークスペースでゲートウェイに移動します。
- UI でイベントを表示するには、 「イベント ログ」 タブをクリックします。
- 詳細な分析を行うには、SQL を使用してイベント ログ テーブルを直接クエリします。
イベントログテーブルをクエリする
flow_progressイベントの行数とバイト数を照会するには：
SELECT
timestamp,
CONCAT(origin.catalog_name, '.', origin.schema_name, '.', origin.dataset_name) AS table_name,
details:flow_progress:metrics:num_upserted_rows::bigint AS rows_upserted,
COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0) AS rows_deleted,
details:flow_progress:metrics:num_output_bytes::bigint AS output_bytes,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'snapshot'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'cdc'
ELSE 'unknown'
END AS ingestion_phase
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND origin.pipeline_type = 'INGESTION_GATEWAY'
ORDER BY timestamp DESC
スナップショットの進行状況パーセンテージを取得するために、 operation_progressイベントを照会するには：
SELECT
timestamp,
origin.flow_name AS flow_name,
details:operation_progress:status::string AS status,
details:operation_progress:progress_percent::double AS progress_pct
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.pipeline_type = 'INGESTION_GATEWAY'
ORDER BY timestamp DESC
<pipeline-id>ゲートウェイ ID に置き換えます。
イベントの構造を理解する
以下の例は、各イベントタイプのJSON構造を示しています。
スナップショットフローの進行状況イベント構造
{
"id": "01234567-89ab-cdef-0123-456789abcdef",
"timestamp": "2025-10-14T13:33:14.175Z",
"level": "METRICS",
"event_type": "flow_progress",
"origin": {
"pipeline_type": "INGESTION_GATEWAY",
"pipeline_name": "MyPipeline",
"dataset_name": "customers",
"catalog_name": "main",
"schema_name": "sales",
"flow_name": "main.sales.customers_snapshot_flow",
"ingestion_source_type": "SQLSERVER"
},
"message": "Completed a streaming update of 'main.sales.customers_snapshot_flow'.",
"details": {
"flow_progress": {
"status": "RUNNING",
"metrics": {
"num_upserted_rows": 7512704,
"num_deleted_rows": null,
"num_output_bytes": 458752000
}
}
},
"maturity_level": "STABLE"
}
CDCフロー進行イベント構造
{
"id": "01234567-89ab-cdef-0123-456789abcdef",
"timestamp": "2025-10-14T13:33:57.426Z",
"level": "METRICS",
"event_type": "flow_progress",
"origin": {
"pipeline_type": "INGESTION_GATEWAY",
"pipeline_name": "MyPipeline",
"dataset_name": "customers",
"catalog_name": "main",
"schema_name": "sales",
"flow_name": "main.sales.customers_cdc_flow",
"ingestion_source_type": "SQLSERVER"
},
"message": "Completed a streaming update of 'main.sales.customers_cdc_flow'.",
"details": {
"flow_progress": {
"status": "RUNNING",
"metrics": {
"num_upserted_rows": 25,
"num_deleted_rows": 3,
"num_output_bytes": 18432
},
"streaming_metrics": {
"discovery_latency_ms": 12450,
"batch_processing_time_ms": 8100,
"event_time": {
"max": "2025-10-14T13:33:45.000Z"
}
}
}
},
"maturity_level": "STABLE"
}
スナップショット操作の進行状況イベント構造
{
"id": "01234567-89ab-cdef-0123-456789abcdef",
"timestamp": "2025-10-14T13:33:14.175Z",
"level": "METRICS",
"event_type": "operation_progress",
"origin": {
"pipeline_type": "INGESTION_GATEWAY",
"pipeline_name": "MyPipeline",
"dataset_name": "customers",
"catalog_name": "main",
"schema_name": "sales",
"flow_name": "main.sales.customers_snapshot_flow",
"ingestion_source_type": "SQLSERVER"
},
"message": "Snapshot in progress for 'main.sales.customers'.",
"details": {
"operation_progress": {
"type": "CDC_SNAPSHOT",
"status": "IN_PROGRESS",
"duration_ms": 3600000,
"progress_percent": 65.5,
"estimated_completion_ms": 1885000,
"cdc_snapshot": {
"target_table_name": "main.sales.customers",
"snapshot_timestamp": 1737542400000,
"snapshot_reason": "NEW_TABLE"
}
}
},
"maturity_level": "STABLE"
}
イベントフィールド
次の表は、進行状況イベントの主要なフィールドについて説明しています。
フィールド | Type | 説明 |
|---|---|---|
| String |
|
| String | 常に |
| String | イベントが生成された際の ISO 8601 タイムスタンプ。 |
| String | 常に |
| String | ゲートウェイの名前。 |
| String | 取り込まれるテーブルの名前。 |
| String | Unity Catalogカタログ名。 |
| String | Unity Catalogスキーマ名。 |
| String | 取り込みフェーズを示すフロー識別子。形式: 初期ロードの場合は |
| String | ソース データベースの種類 (例: |
| String | 現在のフローのステータス。通常は |
| Integer | 前回のイベント以降に挿入または更新された行数。Deltaメトリクス。行とバイトのカウンターを参照してください。 |
| Integer | 前回のイベント以降に削除された行数。Deltaメトリクス。スナップショットフローの |
| Integer | 最後のイベント以降にボリュームにuploadされた圧縮バイト数。Deltaメトリクス。クエリーベースのコネクタスナップショットについては報告されません。クエリーベースのコネクタスナップショットを参照してください。 |
| Integer |
|
| Integer | ゲートウェイが最新の変更バッチの読み取りとアップロードに費やしたミリ秒。ソース側の遅延は含まれません。CDCフローのみです。最初のバッチが完了するまで、 |
| String | このテーブルについてゲートウェイが読み取った最新のソース変更のISO 8601タイムスタンプです。CDCフローのみです。 |
| String | 操作の種類。スナップショット操作の場合は |
| String | 現在の稼働状況。スナップショットが実行されている間 |
| Integer | 操作にかかった合計時間（ミリ秒）。 |
| Double | スナップショットの完了率 ( |
| Integer | スナップショットが終了するまでにかかる推定残り時間（ミリ秒）。スナップショットの進行状況と到着予定時刻を参照してください。 |
| String | スナップショットを作成するテーブルの完全修飾名。 |
| String | 常に |
進捗のメトリクスの解釈
行およびバイトカウンター
num_upserted_rows、num_deleted_rows、および num_output_bytes に適用されます。
- 各値は差分です。これは累積合計ではなく、前のイベントからの変化を表します。合計を計算するには、各イベントの値を合計してください。行数クエリーをご覧ください。
- データに変更がない場合でもイベントは発行されます。ゼロ値は、パイプラインが実行されていることを確認するライブネスシグナルとして機能します。
num_deleted_rowsはスナップショットフローのnullです。スナップショットは削除を生成しないためです。- カウンターは、パイプラインが再起動、更新、または再開されるとResetされます。再起動前に出力されたイベントは、イベントログに残ります。
スナップショットの進捗状況と完了予定時刻
progress_percentおよびestimated_completion_msに適用されます。
progress_percentは累積的です。スナップショットのライフサイクル全体にわたって0.0から100.0まで累積します。progress_percentは(completed_chunks / total_chunks) × 100として計算されるため、正確な行レベルのパーセンテージではなく、概算です。複数のチャンクに分割されていないテーブル（通常は小さいテーブル）は、0.0から直接100.0にジャンプします。多くのチャンクに分割される大きなテーブルは、各チャンクの完了に伴い徐々に更新されます。COMPLETEDステータスはprogress_percent = 100.0を報告します。- パイプラインを再起動すると、
progress_percentを含む進捗メトリクスがReset される可能性があります。 estimated_completion_msイベントが発行された時点での見積もりを反映します。スナップショットの進行に伴って減少し、完了すると0になります。進捗が遅くなると一時的に増加する可能性があり、見積もりを作成するのに十分なデータが処理されるまでnullとなります。
CDCレイテンシ
event_time:max、 discovery_latency_ms、および batch_processing_time_msに適用されます。これらのメトリクスは、CDCフローに対してのみ報告されます。
- 各値は、イベントが発行された瞬間のメトリクスの状態を反映しています。
- これらは、インジェクションゲートウェイの遅延です。これらは、ソースデータベースからゲートウェイを経由して Unity Catalog ボリュームに至るパスを測定します。これらには、Unity Catalog ボリュームから宛先テーブルまでのダウンストリームの遅延は含まれません。これらはインジェクションパイプラインのイベントログで別途報告されます。
- ソース側のラグを推定するには、
discovery_latency_msからbatch_processing_time_msを減算します。 discovery_latency_msまた、結果が負になる場合は、batch_processing_time_msは0として報告されます。
これらのメトリクスの関係から、ゲートウェイパスのどの地点でレイテンシが集中しているかがわかります：
パターン | 意味 |
|---|---|
| CDC は実行中で、最新の状態です。 |
| ソース側の遅延。ゲートウェイは速やかに読んでいます。ソースからコミットが遅延しました（レプリケーションの遅延、ソースログのバックログ、実行時間の長いソーストランザクション）。 |
両方とも高い | ゲートウェイ側の遅延。アップロードパイプラインはボトルネックです。ゲートウェイコンピュートリソースおよびUnity Catalogボリュームへのネットワークパスを確認してください。 |
| テーブルごとのソースの問題（DDL処理中、ロック競合、ブロックされたレプリケーションスロット、スキーマ変更） |
| ゲートウェイ全体での問題（リソース、ボリューム接続、ソース CDC ログバックログ、すべてのキャプチャに影響） |
| ソースはアイドル状態です、またはゲートウェイがソースデータベースへの接続を失いました。ソース CDC の有効化とゲートウェイログを確認してください。 |
クエリーベースのコネクタスナップショット
ゲートウェイがクエリーベースのコネクタ（QBC）メソッドを使用して初期スナップショットを実行すると、このページで説明されているのと同じスナップショットメトリクスが出力されます。行数には flow_progress イベント、進捗率と推定残り時間 (estimated_completion_ms) には operation_progress イベントが使用されます。
これらのメトリクスは、取り込みゲートウェイからのみ取得されます。ゲートウェイなしで実行されるクエリーベースのコネクタパイプライン（接続を介してソースを直接読み取るパイプライン）はそれらを出力しないため、そのイベントlogsにはスナップショットの進捗イベントが含まれません。
QBC スナップショットは、以下の点でゲートウェイ スナップショット（レガシー）と異なります。
- 出力バイト数は報告されません。
flow_progressイベントは行数（num_upserted_rows）を報告しますが、num_output_bytesは報告しません。出力バイトクエリーにあるような、出力バイトを読み取るクエリーは、QBCスナップショットのデータを返しません。代わりに、行数メトリクスを使用してボリュームを追跡してください。 - テーブルは
SCHEDULEDステータスを報告します。 各テーブルはスナップショット開始前にキュー中にSCHEDULEDを報告します。ゲートウェイのスナップショット(レガシー)はキュー状態を報告しません。 - 一部の
operation_progressフィールドが入力されていません。duration_msフィールドとcdc_snapshot.snapshot_reasonフィールドは、QBC スナップショットには設定されていません。
QBCスナップショットを監視するには、スナップショット進捗クエリーでスナップショットの進行状況クエリーを使用し、推定残余時間のテーブルごとのスナップショットETAクエリーも含めて使用してください。
QBCスナップショットがキュー中のテーブルをどのように報告するかのクエリーを調整してください。そのセクションのクエリーは、 0の進行率でキュー中のテーブルを識別しますが、QBCスナップショットは SCHEDULED ステータスのテーブルを報告し、進捗率は報告しません。正しく分類またはカウントするには、以下の条件を変更してステータスを確認してください。
progress_pct = 0条件をstatus = 'SCHEDULED'に置き換えます。- 「テーブルごとのスナップショット ETA」クエリーにキューに入れられたテーブルを含めるには、その
status = 'IN_PROGRESS'フィルターをstatus IN ('IN_PROGRESS', 'SCHEDULED')に変更します。
サンプルクエリ
以下のサンプルクエリは、行数、出力バイト数、スナップショットの進捗状況とETA、およびCDCレイテンシーメトリクスを使用してゲートウェイを監視する方法を示しています。各クエリで、<pipeline-id>をゲートウェイIDに置き換えます。
取り込みゲートウェイ プログレスモニターノートブックには、このセクションのすべてのクエリーが含まれています。ゲートウェイが実行されているワークスペースにノートブックをインポートし、ゲートウェイ ID を指定してください。スナップショットの進捗、CDCレイテンシー、および行数を検査するためにノートブックを実行してください。モニタリング要件に合わせてSLAしきい値を調整することもできます。
行数クエリ
テーブルごとの売上高（過去24時間）
過去24時間における各テーブルのアップサートと削除の合計数。取り込みフェーズはフロー名の接尾辞から分類されます。これをダッシュボードの見出しとして使用すれば、どのテーブルが最も多くのデータを移動したかを確認できます。
WITH row_events AS (
SELECT
origin.flow_name AS flow_name,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
ELSE 'OTHER'
END AS phase,
details:flow_progress:metrics:num_upserted_rows::bigint AS upserts,
details:flow_progress:metrics:num_deleted_rows::bigint AS deletes
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
)
SELECT
flow_name,
phase,
SUM(upserts) AS rows_upserted_24h,
SUM(COALESCE(deletes, 0)) AS rows_deleted_24h,
SUM(upserts) + SUM(COALESCE(deletes, 0)) AS total_rows_moved_24h
FROM row_events
GROUP BY flow_name, phase
ORDER BY total_rows_moved_24h DESC
サイレントテーブルまたはスタックテーブルを特定する
イベントを発信しているにもかかわらず、過去60分間にアップサートと削除がゼロであると報告しているテーブルは、「スタック」状態になる可能性が高い。番号を変更する必要があるかどうか、さらに調査してください。 CDCの表のうち、実際に利用されていないもの（例えば、夜間など）もここに表示されます。
WITH recent_window AS (
SELECT
origin.flow_name AS flow_name,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
ELSE 'OTHER'
END AS phase,
COUNT(*) AS emissions_in_window,
SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS upserts_in_window,
SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0)) AS deletes_in_window,
MAX(timestamp) AS last_event
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND timestamp >= current_timestamp() - INTERVAL 60 MINUTES
GROUP BY origin.flow_name
)
SELECT
flow_name,
phase,
emissions_in_window,
upserts_in_window,
deletes_in_window,
last_event,
ROUND(TIMESTAMPDIFF(MINUTE, last_event, current_timestamp()), 0) AS minutes_since_last_event
FROM recent_window
WHERE upserts_in_window = 0
AND deletes_in_window = 0
ORDER BY minutes_since_last_event DESC
テーブルごとのタイムラインと累計値
過去24時間における、単一のフローに関するイベントごとの完全な履歴と、累積行数。<flow-pattern> SQL LIKEパターン (例: '%customers%_cdc_flow' ) に置き換えます。
SELECT
origin.flow_name AS flow_name,
origin.update_id AS update_id,
timestamp,
details:flow_progress:metrics:num_upserted_rows::bigint AS upserts_this_period,
COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0) AS deletes_this_period,
SUM(details:flow_progress:metrics:num_upserted_rows::bigint)
OVER (PARTITION BY origin.flow_name, origin.update_id
ORDER BY timestamp
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cumulative_upserts_this_run,
SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0))
OVER (PARTITION BY origin.flow_name, origin.update_id
ORDER BY timestamp
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cumulative_deletes_this_run
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND origin.flow_name LIKE '<flow-pattern>'
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY timestamp
出力バイトクエリ
テーブルあたりのバイト数（過去24時間）
過去24時間にテーブルごとにボリュームにアップロードされた合計バイト数（MBおよびGB単位）。最もデータ量の多いテーブルを表示するには、降順で並べ替えてください。
WITH byte_events AS (
SELECT
origin.flow_name AS flow_name,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
ELSE 'OTHER'
END AS phase,
details:flow_progress:metrics:num_output_bytes::bigint AS output_bytes
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
)
SELECT
flow_name,
phase,
SUM(output_bytes) AS bytes_24h,
ROUND(SUM(output_bytes) / 1024.0 / 1024.0, 2) AS mb_24h,
ROUND(SUM(output_bytes) / 1024.0 / 1024.0 / 1024.0, 3) AS gb_24h
FROM byte_events
GROUP BY flow_name, phase
ORDER BY bytes_24h DESC
スループットの傾向（MB/分）
過去24時間における全フローでのアップロードバイト数の、1分ごとの時系列データ。スループットのパターンや停滞箇所を特定するために、折れ線グラフとして表示します。
SELECT
DATE_TRUNC('MINUTE', timestamp) AS ts_minute,
origin.flow_name AS flow_name,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
ELSE 'OTHER'
END AS phase,
ROUND(SUM(details:flow_progress:metrics:num_output_bytes::bigint) / 1024.0 / 1024.0, 2) AS mb_per_minute
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY DATE_TRUNC('MINUTE', timestamp), origin.flow_name
ORDER BY origin.flow_name, ts_minute
行あたりの平均バイト数（ワイドテーブルまたはLOB検出器）
num_output_bytesと行数を結合して、テーブルごとの行ごとの平均バイト数を計算します。 高い値は通常、スループットコストを押し上げるLOBテーブルまたはワイドスキーマテーブルを示しています。キャパシティプランニングやスキーマレビューに役立ちます。
WITH joined AS (
SELECT
origin.flow_name AS flow_name,
CASE
WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
ELSE 'OTHER'
END AS phase,
SUM(details:flow_progress:metrics:num_output_bytes::bigint) AS total_bytes,
SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS total_upserts,
SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0)) AS total_deletes
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY origin.flow_name
)
SELECT
flow_name,
phase,
total_upserts + total_deletes AS total_rows,
ROUND(total_bytes / 1024.0 / 1024.0, 2) AS total_mb,
ROUND(total_bytes / NULLIF(total_upserts + total_deletes, 0), 0) AS avg_bytes_per_row
FROM joined
WHERE total_bytes > 0
ORDER BY avg_bytes_per_row DESC
スナップショットの進捗状況に関するクエリ
全体的なスナップショットの進捗状況
次のクエリは、完了、進行中、またはキューにあるテーブルの数の単一行の概要を返します。結果には、イベントログの保持期間内におけるすべてのパイプライン更新を通じて、テーブルごとに報告された最新の状態が反映されます。そのため、以前の更新で完了したテーブルは、更新または再起動後も引き続きtables_completedにカウントされます。
WITH latest_per_table AS (
SELECT
origin.flow_name AS flow_name,
details:operation_progress:status::string AS status,
details:operation_progress:progress_percent::double AS progress_pct,
ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
)
SELECT
COUNT(*) AS total_tables,
SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) AS tables_completed,
SUM(CASE WHEN status = 'IN_PROGRESS' AND progress_pct > 0 AND progress_pct < 100 THEN 1 ELSE 0 END) AS tables_in_progress,
SUM(CASE WHEN progress_pct = 0 THEN 1 ELSE 0 END) AS tables_not_started,
ROUND(AVG(progress_pct), 2) AS overall_progress_pct,
ROUND(SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) AS pct_tables_done
FROM latest_per_table
WHERE rn = 1
テーブルごとのスナップショットステータスボード
以下のクエリは、パイプライン内の各テーブルと、その最新の報告ステータスおよび進捗率を返します。結果には、以前の更新で完了したテーブルが含まれます。そのため、欠落している行として表示されるテーブルはありません。
WITH table_status AS (
SELECT
origin.flow_name AS flow_name,
details:operation_progress:status::string AS status,
details:operation_progress:progress_percent::double AS progress_pct,
timestamp,
ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
)
SELECT
flow_name,
status,
ROUND(progress_pct, 2) AS progress_pct,
timestamp AS last_update,
CASE
WHEN status = 'COMPLETED' THEN 'Done'
WHEN progress_pct = 0 THEN 'Queued'
ELSE 'Active'
END AS phase
FROM table_status
WHERE rn = 1
ORDER BY
CASE status WHEN 'IN_PROGRESS' THEN 0 WHEN 'COMPLETED' THEN 1 ELSE 2 END,
progress_pct ASC
現在の更新で読み込まれたスナップショットの行数とバイト数
次のクエリは、現在のスナップショット実行における各テーブルの進捗率、アップサートされた行、およびアップロードされたバイト数を組み合わせています。
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC LIMIT 1
),
progress AS (
SELECT
origin.flow_name AS flow_name,
details:operation_progress:progress_percent::double AS progress_pct,
details:operation_progress:status::string AS status,
ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.update_id = (SELECT update_id FROM latest_update)
),
volume AS (
SELECT
origin.flow_name AS flow_name,
SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS rows_loaded,
SUM(details:flow_progress:metrics:num_output_bytes::bigint) AS bytes_loaded
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.flow_name LIKE '%_snapshot_flow'
AND origin.update_id = (SELECT update_id FROM latest_update)
GROUP BY origin.flow_name
)
SELECT
p.flow_name,
p.status,
ROUND(p.progress_pct, 2) AS progress_pct,
v.rows_loaded,
ROUND(v.bytes_loaded / 1024.0 / 1024.0, 2) AS mb_loaded,
ROUND(v.bytes_loaded / 1024.0 / 1024.0 / 1024.0, 3) AS gb_loaded
FROM progress p
LEFT JOIN volume v ON p.flow_name = v.flow_name
WHERE p.rn = 1
ORDER BY p.progress_pct ASC
停止したスナップショットの検出
次のクエリは、過去30分間にprogress_percentが変更されていないスナップショットテーブルを返します。停止しているが、まだアクティブであるスナップショットを特定するために使用します。
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC LIMIT 1
),
recent AS (
SELECT
origin.flow_name AS flow_name,
details:operation_progress:progress_percent::double AS progress_pct,
details:operation_progress:status::string AS status,
timestamp
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.update_id = (SELECT update_id FROM latest_update)
AND timestamp >= current_timestamp() - INTERVAL 30 MINUTES
)
SELECT
flow_name,
ROUND(MIN(progress_pct), 2) AS min_pct_30min,
ROUND(MAX(progress_pct), 2) AS max_pct_30min,
ROUND(MAX(progress_pct) - MIN(progress_pct), 2) AS pct_change_30min,
COUNT(*) AS events_in_window,
MAX(timestamp) AS last_event_ts
FROM recent
WHERE status = 'IN_PROGRESS'
GROUP BY flow_name
HAVING MAX(progress_pct) - MIN(progress_pct) = 0
AND MAX(progress_pct) < 100
ORDER BY max_pct_30min ASC
テーブルごとのスナップショット完了予定時刻
以下のクエリは、アクティブなスナップショット実行における各テーブルの現在の進捗率および完了までの推定時間を返します。IN_PROGRESS のステータスのテーブルのみが返されます。完了済みおよびキューにあるテーブルを含めるには、status = 'IN_PROGRESS'フィルターを削除します。
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC LIMIT 1
),
latest_per_flow AS (
SELECT
origin.flow_name AS flow_name,
details:operation_progress:status::string AS status,
details:operation_progress:progress_percent::double AS progress_pct,
details:operation_progress:estimated_completion_ms::bigint AS eta_ms,
timestamp,
ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.update_id = (SELECT update_id FROM latest_update)
)
SELECT
flow_name,
status,
ROUND(progress_pct, 2) AS progress_pct,
eta_ms,
ROUND(eta_ms / 1000.0 / 60.0, 1) AS eta_minutes,
ROUND(eta_ms / 1000.0 / 3600.0, 2) AS eta_hours,
timestamp AS last_update
FROM latest_per_flow
WHERE rn = 1
AND status = 'IN_PROGRESS'
ORDER BY eta_ms DESC NULLS LAST
CDC レイテンシークエリ
これらのクエリーが返す値を解釈するには、 CDCのレイテンシーを参照してください。
CDCのテーブルごとの最新性
次のクエリーは、各CDCテーブルの直近30分間のCDC鮮度イベントを返します。5分おきにイベントが送信されるため、テーブルごとに約6行が生成されます。結果はdiscovery_latency_msでソートされているため、最も遅延しているテーブルが先に表示されます。
SELECT
origin.flow_name AS flow_name,
details:flow_progress:streaming_metrics:event_time:max::string AS latest_source_commit_seen,
details:flow_progress:streaming_metrics:discovery_latency_ms::bigint AS discovery_latency_ms,
details:flow_progress:streaming_metrics:batch_processing_time_ms::bigint AS batch_processing_time_ms,
timestamp AS emission_ts
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.flow_name LIKE '%_cdc_flow'
AND details:flow_progress:streaming_metrics IS NOT NULL
AND timestamp >= current_timestamp() - INTERVAL 30 MINUTES
ORDER BY discovery_latency_ms DESC NULLS LAST, flow_name, emission_ts DESC
CDC レイテンシ時系列
以下のクエリは、直近24時間における各CDCフローのdiscovery_latency_msとbatch_processing_time_msの時間平均を返します。鮮度が低下した時期を特定するためにご利用ください。多数のCDCフローを含むパイプラインでは、結果セットを絞り込むには、フロー名でフィルタリングしてください。
SELECT
DATE_TRUNC('HOUR', timestamp) AS ts_hour,
origin.flow_name AS flow_name,
ROUND(AVG(details:flow_progress:streaming_metrics:discovery_latency_ms::bigint) / 1000.0, 2) AS avg_discovery_latency_sec,
ROUND(AVG(details:flow_progress:streaming_metrics:batch_processing_time_ms::bigint) / 1000.0, 2) AS avg_batch_processing_sec
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
AND level = 'METRICS'
AND origin.flow_name IS NOT NULL
AND origin.pipeline_type = 'INGESTION_GATEWAY'
AND origin.flow_name LIKE '%_cdc_flow'
AND details:flow_progress:streaming_metrics IS NOT NULL
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY DATE_TRUNC('HOUR', timestamp), origin.flow_name
ORDER BY origin.flow_name, ts_hour
トラブルシューティング
進行イベントが表示されない
イベント ログに進行状況イベントが表示されない場合は、次の手順を実行します。
- パイプラインが起動してから少なくとも5分間お待ちください。
- パイプラインがアクティブに実行され、取り込まれていることを確認します。
- クエリーが
level = 'METRICS'と正しいイベント タイプ (flow_progressまたはoperation_progress) でフィルター処理されていることを確認してください。 - イベントがまだ欠落している場合、または5分以外の間隔でイベントが到着する場合は、Databricksサポートにお問い合わせください。
パイプラインの再起動後にメトリクスがゼロになる
パイプラインの再起動、更新、または再開により、進捗メトリクスがResetされる場合があります。ゲートウェイはすぐに新しい値のレポートを開始し、再起動前に出力されたイベントはイベントログに残ります。進捗メトリクスの解釈を参照してください。
一部のテーブルにメトリクスがありません
一部のテーブルに進行状況イベントが表示されない場合:
- テーブルがゲートウェイ設定に含まれており、フィルタで除外されていないことを確認してください。
- CDCフェーズでは、ソーステーブルでCDCまたは変更追跡が有効になっていることを確認してください。
operation_progressイベントはスナップショットフローに対してのみ発行されることに注意してください。CDCには完了の概念がないため、CDCフローはoperation_progressイベントを発行しません。
スナップショットの推定到着時刻（ETA）またはCDCレイテンシのフィールドは null
一部のフィールドは、Gatewayが報告に必要な十分な情報を持つまでnullとなります。各フィールドがnullの場合の詳細は、イベントフィールドと進行状況メトリクスの解釈を参照してください。