チェンジデータキャプチャ（CDC）テーブルのデータの鮮度を監視する
適用対象 : 
データベースコネクタ
プレビュー
この機能は パブリック プレビュー段階です。
この機能は順次リリースされます。パイプラインに データの鮮度 列が表示されない場合、その機能はワークスペースで利用できません。
データの鮮度は、各宛先テーブルのデータがソースデータベースと比較してどの程度最新であるかを示します。データベースコネクタパイプラインの各テーブルについて、パイプライン UI は、パイプラインが取り込んだ直近のソース変更から経過した時間を表示します。
データの鮮度は、SQL Server、PostgreSQL、MySQL、Oracleソースで利用できます。
データの鮮度を使用して、以下の操作を実行します。
- イベントログを照会せずに、各宛先テーブルがソースよりどれだけ遅れているかを確認します。
- パイプラインの他の部分に遅れているテーブルを特定します。
- 変更を積極的に受信しているテーブルと、ソースでアイドル状態のテーブルを区別します。
データの鮮度の計算方法
データの鮮度は、次の値から各テーブルについて計算されます。
Value | 定義 |
|---|---|
ソース commit 時間 | トランザクションがソースデータベースでコミットされた時刻。 |
最新のソースcommit時刻 | パイプラインがテーブルに取り込んだ変更の中で、最新のソースcommit時間。 |
データの鮮度 | 現在時刻から最新のソース commit 時刻を引いたもの。 |
たとえば、テーブルに取り込まれた最新の変更が3分前にソースでコミットされた場合、テーブルのデータの鮮度は3mになります。値が小さいほど、宛先テーブルがソースの現在の状態に近いことを意味します。
データ鮮度は、出力先テーブルがどれくらい最新であるかを測定します。これは、ゲートウェイのイベントログにあるdiscovery_latency_msのようなCDCレイテンシーメトリクスとは異なります。これらのメトリクスは、ソースデータベースからステージングボリュームへのパスのみを測定します。CDCレイテンシーを参照してください。
データの鮮度を表示する
- Databricks ワークスペースで、サイドバーの ジョブ & パイプライン をクリックします。
- 取り込みパイプラインの名前をクリックします。
- テーブルのリストで、 データの鮮度 列を探します。
この列には、各テーブルの値が時間、分、秒で表示されます（例：3m 23sまたは45h 32m 4s）。
データの鮮度の動作
- 実行中の更新 : データの鮮度が継続的に更新されます。この値は時間が経つにつれて増加し、パイプラインがソースから新しい変更を取り込むと減少します。
- 停止または完了した更新 ：更新が終了すると、データの鮮度の更新は停止します。列は更新が終了した時点の値を表示し続けるため、パイプラインが停止した後でも各テーブルの鮮度を確認できます。
- アイドル状態のテーブル : データの鮮度は、パイプラインがソースから受け取った最後の変更を反映します。ソーステーブルに新しいcommitがない場合、パイプラインが正常で完全に追いついている場合でも、そのデータの鮮度は低下し続けます。
データの鮮度を解釈する
アイドル状態のソーステーブルもデータの鮮度の向上を示すため、遅延として処理する前にその値をソースのアクティビティと比較してください。
遅延の原因を特定する
インジェクションゲートウェイを使用するパイプラインの場合、ゲートウェイが読み取った最新のソース変更と、インジェクションパイプラインがインジェストした最新のソース commit 時間を比較します。ゲートウェイは、そのイベントログで event_time:max をレポートします。インジェクションパイプラインは、そのイベントログで upstream_time:max をレポートします。「イベント ログからのデータの鮮度のクエリー」を参照してください。
ゲートウェイ | 取り込みパイプライン | 意味 |
|---|---|---|
進行中 | 進行中 | 変更はパイプラインを流れています。 |
進行中 | 進んでいません | 取り込みパイプラインはゲートウェイの背後にあります。取り込みパイプラインの状態とそのスケジュールを確認します。 |
進んでいません | 進んでいません | ソースがアイドル状態か、ゲートウェイがソースへの接続を失っています。ゲートウェイを診断するには、CDCレイテンシを参照してください。 |
イベントログからのデータ鮮度のクエリー
データの鮮度の基礎となるメトリクスは、ゲートウェイのイベントログではなく、インジェクションパイプラインのイベントログに記録されます。イベントログをクエリーして、時間の経過に伴うデータの鮮度を追跡し、テーブルを比較し、アラートを作成します。
CDCフローの各 flow_progress イベントは、 details:flow_progress:streaming_metricsの下で次のフィールドを報告します。
フィールド | 型 | 説明 |
|---|---|---|
| String | バッチ内の変更の中で最新のソースcommit時刻。ISO 8601 Timestamp (UTC) として表されます。 |
| String | バッチ内の変更の中で最も早いソースcommit時刻（UTCでのISO 8601 Timestamp）。 |
| Integer | バッチのエンドツーエンドの遅延（ミリ秒単位）。遅延は、バッチ内の最も古い変更のソースcommit時間から、バッチがターゲットテーブルにcommitされたときまでに測定されます。 |
各クエリーで、<ingestion-pipeline-id>をインジェクションパイプラインのIDに置き換えます。ゲートウェイのイベント Logsには、これらのフィールドはレポートされません。
テーブルごとのデータの鮮度と最新のラグ
次のクエリーは、各テーブルの最新のソース commit 時間、データの鮮度、最後のバッチ時間、および最新のバッチのエンドツーエンドのラグを返します。結果は、最も鮮度が低いテーブルが最初に表示されるようにソートされます。
クエリーには、パイプラインの最新の更新にあるテーブルのみが含まれます。このフィルターがないと、パイプラインから削除したテーブルがイベントログに残り、それらのデータの鮮度が増加し続けます。
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
)
SELECT
origin.flow_name AS flow_name,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp) AS latest_source_commit_time,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes,
MAX(timestamp) AS last_batch_time,
MAX_BY(details:flow_progress:streaming_metrics:stream_latency_ms::bigint, timestamp) AS latest_lag_ms
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
ORDER BY data_freshness_minutes DESC
時間経過に伴うエンドツーエンドのラグ
以下のクエリーは、過去24時間の各テーブルの1時間あたりの最大エンドツーエンドラグを返します。これを使用して、ラグがいつ増加したかを見つけます。
SELECT
DATE_TRUNC('HOUR', timestamp) AS ts_hour,
origin.flow_name AS flow_name,
ROUND(MAX(details:flow_progress:streaming_metrics:stream_latency_ms::bigint) / 1000.0 / 60.0, 1) AS max_lag_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND details:flow_progress:streaming_metrics:stream_latency_ms 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
データの鮮度に関するアラートを作成する
テーブルの遅延が発生したときに通知を受け取るには、データの鮮度クエリーでDatabricks SQLアラートを作成します。たとえば、次のクエリーは、データの鮮度が120分を超えているパイプラインの最新の更新にあるテーブルの数を返します。120をthresholdに置き換え、 stale_tablesが 0より大きい場合にTriggerされるようにアラートを設定します。
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
),
per_table_freshness AS (
SELECT
origin.flow_name AS flow_name,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
)
SELECT
COUNT_IF(data_freshness_minutes > 120) AS stale_tables
FROM per_table_freshness
データの鮮度値がまだないテーブル（初期スナップショットが実行されているテーブルなど）はカウントされません。
クエリーは最新のソース commit 時刻と現在時刻を比較するため、アイドル状態のソーステーブルや、パイプラインが実行されていない間も値は増え続けます。UIに表示される値とは異なり、更新が終了しても変化は止まりません。データが最新のときにアラートが発生するのを防ぐには：
- ソーステーブルが変更を受け取る頻度に基づいてthresholdを設定します。
- スケジュールに基づいて実行されるパイプラインの場合、スケジュールされたランの間隔よりも高いthresholdを設定します。
- テーブルの変更頻度が大きく異なる場合は、変更頻度が同程度のテーブルのグループごとにアラートを個別に作成し、それぞれのクエリーに
flow_nameでフィルターを適用してください。
Databricks SQL のアラートとアラートを作成するをご覧ください。
制限
- 変更のないテーブル : パイプラインが起動してからテーブルに変更がない場合、データの鮮度は空になります。UIにはダッシュ（
-）が表示され、イベントログにはテーブルのupstream_time値がありません。これには、初期スナップショットが実行されているテーブルが含まれます。 - SQL Server ソース : 変更のソース commit 時間が利用できない場合、ゲートウェイは 120 秒ごとに更新するキャッシュされた commit 時間を使用します。キャッシュされた値が更新されるまで、データの鮮度は実際の遅延より高い場合があります。
- Oracleソース ：
BLOB、CLOB、NCLOBなどのラージオブジェクト（LOB）列を持つOracleテーブルの場合、データの鮮度が負になる可能性があります。負の値は、記録されたソース commit 時間が現在の時間より後であることを意味します。これは、データがソースよりも新しいことを意味するものではありません。負の値が表示される場合は、Databricksサポートに問い合わせてください。