パイプライン イベントシステムテーブルリファレンス

この記事は、アカウント内のpipeline_events イベント ログ エントリを記録する、 システム テーブルのリファレンスです。 各行は、パイプライン イベント ログからの不変イベントであり、ライフサイクルの遷移、フローの進行状況、データ品質メトリック、エラー、クラスター リソース、およびリージョン内のすべてのパイプラインとワークスペース全体のその他の運用データをキャプチャします。

必要条件

• このシステム テーブルにアクセスするには、ユーザーは次のいずれかを行う必要があります。
  • メタストア管理者とアカウント管理者の両方であるか、または
  • システム スキーマに対する USE 権限と SELECT 権限を持つ。 システム テーブルへのアクセス権の付与を参照してください。

使用可能なパイプライン イベント テーブル

パイプラインイベントシステムのテーブルはベータ版中は lakeflow_pipeline_events_preview スキーマに存在し、一般公開時に lakeflow スキーマに移動します。

詳細なスキーマ リファレンス

パイプライン イベント テーブル スキーマ

パイプライン イベント テーブルは追加専用です。 各行は、パイプラインの更新によって生成された 1 つのイベントを出力時に記録し、行が変更されたり削除されたりすることはありません。

行に入力されるフィールドは、イベントの種類によって異なります。 error、 update_id、および多くの origin.* サブフィールドは、それらが適用されるイベントにのみ設定され、 details フィールドの構造も event_typeによって異なります。

このテーブルを使用して、履歴パイプライン アクティビティのクエリを実行し、パイプラインエラーに関するアラートを作成し、パイプラインの動作を他の Lakeflow システム テーブルと関連付けます。

テーブル パス: system.lakeflow_pipeline_events_preview.pipeline_events

主キー: (account_id, pipeline_event_id)

起源構造フィールド

エラー構造フィールド

詳細分野

details列はVARIANTであり、含まれるフィールドはevent_typeによって異なります。 各イベントの種類で使用できるフィールドについては、「 パイプライン イベント ログ スキーマ」を参照してください。 variant_get関数またはドット構文を使用して、入れ子になった値を読み取ります。 一般的なアクセス パターンについては、以下のクエリ例を参照してください。

event_typeキーがペイロードを巻きつけます。 例えば、 flow_progress イベントの指標は $.flow_progress.metricsであり、 $.metricsではありません。 すべてのパスにイベントタイプキーを含めてください。

-- Using variant_get (lets you cast to a specific type)
SELECT
  pipeline_id,
  event_time,
  variant_get(details, '$.flow_progress.status', 'STRING')                       AS flow_status,
  variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')      AS rows_written,
  variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')        AS backlog_bytes
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

-- Using dot syntax (returns VARIANT, cast when needed)
SELECT
  pipeline_id,
  event_time,
  details:flow_progress.status::STRING               AS flow_status,
  details:flow_progress.metrics.num_output_rows::BIGINT AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

クエリの例

-- Flow throughput for a specific pipeline
SELECT
  origin.flow_name,
  date_trunc('HOUR', event_time) AS hour,
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  pipeline_id = '<your-pipeline-id>'
  AND event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  origin.flow_name,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC,
  rows_written DESC

-- The latest error for each pipeline that has errored in the last 7 days, with the outermost exception.
-- The exception chain is ordered with the root cause last, so read element -1 for the root cause.
-- On many errors only the first element carries error_class and sql_state.
SELECT
  workspace_id,
  pipeline_id,
  event_time,
  event_type,
  message,
  error.exceptions[0].error_class AS exception_error_class,
  error.exceptions[0].sql_state   AS exception_sql_state
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
QUALIFY
  ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY event_time DESC) = 1
ORDER BY
  event_time DESC

-- Data quality: failed expectations by dataset, per update, in the last 1 day
SELECT
  pipeline_id,
  update_id,
  origin.dataset_name,
  expectation.name AS expectation_name,
  SUM(expectation.failed_records) AS failed_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
  LATERAL VIEW explode(variant_get(details, '$.flow_progress.data_quality.expectations', 'ARRAY<STRUCT<name:STRING,dataset:STRING,passed_records:BIGINT,failed_records:BIGINT>>')) AS expectation
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 DAY
GROUP BY
  pipeline_id,
  update_id,
  origin.dataset_name,
  expectation.name
HAVING
  SUM(expectation.failed_records) > 0
ORDER BY
  failed_records DESC

一般的な結合パターン

pipelines テーブルと結合してパイプライン名でフィルター処理する

pipelines テーブルは、緩やかに変化するディメンション (SCD2) です。 参加する前に、各パイプラインの最新バージョンを取得します。

WITH latest_pipelines AS (
  SELECT *
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC) = 1
)
SELECT
  p.name AS pipeline_name,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events e
JOIN
  latest_pipelines p
  ON e.workspace_id = p.workspace_id
  AND e.pipeline_id = p.pipeline_id
WHERE
  e.level = 'ERROR'
  AND e.event_time >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY
  e.event_time DESC

pipeline_update_timelineを使用して参加するupdate_id

SELECT
  u.period_start_time AS update_start,
  u.period_end_time   AS update_end,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  system.lakeflow.pipeline_update_timeline u
JOIN
  system.lakeflow_pipeline_events_preview.pipeline_events e
  ON e.update_id = u.update_id
WHERE
  u.pipeline_id = '<your-pipeline-id>'
  AND u.period_start_time >= current_timestamp() - INTERVAL 7 DAYS
ORDER BY
  u.period_start_time DESC,
  e.event_time ASC

アラートの設定

pipeline_events アラートを使用して、にアラートを作成できます。 pipeline_events (必要に応じて他の Lakeflow システム テーブルと結合) に対して SQL クエリを作成し、SQL ウェアハウスでスケジュールを設定し、通知先 (電子メール、Slack、webhook、PagerDuty) を構成します。

いくつかの便利な開始点:

過去 N 分間にパイプラインに対してイベントが到着しなかった場合のアラート

これを使用して、スタックまたはサイレントで失敗したパイプラインを検出します。

-- Returns one row per pipeline that has not emitted any event in the last 30 minutes.
-- The alert can trigger when this query returns any rows.
SELECT
  workspace_id,
  pipeline_id,
  MAX(event_time) AS last_event_time
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_time >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY
  workspace_id,
  pipeline_id
HAVING
  MAX(event_time) < current_timestamp() - INTERVAL 30 MINUTES

特定のフローのバックログが高すぎる場合にアラートを生成する

バックログはflow_progressbacklog_bytesイベントごとに報告され、ファイルソースについてはbacklog_filesとして報告されます。 直近の読み取り値が閾値を超えたとき(例えば、未処理の作業が100MB分)発生したときにトリガーされます。 すべての情報源がすべての指標を報告しているわけではないので、情報源が入力されている指標でフィルタリングしてください。

-- Returns the most recent backlog reading per flow for a given pipeline.
-- The alert can trigger when backlog_bytes exceeds the threshold for any flow.
WITH latest_flow_progress AS (
  SELECT
    workspace_id,
    pipeline_id,
    origin.flow_name,
    event_time,
    CASE
      WHEN variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED' THEN 0
      ELSE variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')
    END AS backlog_bytes
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    pipeline_id = '<your-pipeline-id>'
    AND event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
    AND (
      variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT') IS NOT NULL
      OR variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED'
    )
  QUALIFY
    ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id, origin.flow_name ORDER BY event_time DESC) = 1
)
SELECT *
FROM latest_flow_progress
WHERE backlog_bytes > 100000000

どのソースが遅れているかを見るために、$.flow_progress.metrics.source_metricsは各ソースごとの読み取り値の配列で、それぞれのソースの測定値の横にそのソースの1source_name、backlog_bytes、backlog_recordsがbacklog_files付きです。

パイプラインでのデータ品質の低下に関するアラート

各 flow_progress イベントは、予想される EXPECT … DROP 削除された行の数を報告します。 これらを更新ウィンドウでデータセットごとに合計し、合計が閾値を超えたらアラートします。

-- Returns datasets where more than 100 rows were dropped by expectations, per update, in the last hour.
SELECT
  pipeline_id,
  update_id,
  origin.dataset_name,
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) AS dropped_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR
GROUP BY
  pipeline_id,
  update_id,
  origin.dataset_name
HAVING
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) > 100

フローが処理する行数が少なすぎるとアラートします

flow_progress イベントは metrics.num_output_rows マイクロバッチごとのカウントとして報告されるため、ウィンドウ内のイベントを合計すると、そのウィンドウに書き込まれた行が得られます。 スループットが予想される下限を下回ったときにアラートを作成してください。 例えば、通常は1時間に何千行も書き込むフローがほぼゼロになる場合、ソースの設定が誤っていることを示しています。

このクエリは、ウィンドウ内に行数のある flow_progress イベントを発生させたフローのみを報告します。 完全に停止したフローはイベントを発生させないため、このアラートと上記の欠損イベントアラートを組み合わせてください。

-- Returns flows that wrote fewer than 100 rows in the last hour.
-- The alert can trigger when this query returns any rows.
SELECT
  pipeline_id,
  origin.flow_name,
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) AS rows_written_last_hour
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  pipeline_id = '<your-pipeline-id>'
  AND event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR
  AND variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT') IS NOT NULL
GROUP BY
  pipeline_id,
  origin.flow_name
HAVING
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) < 100

新規データ遅延が高すぎるとアラート

ストリーミングフローの場合、 flow_progress イベントは streaming_metricsの遅延を報告します。 stream_latency_ms はデータが上流に到達してからマイクロバッチがデルタテーブルにコミットするまでの時間です。 直近の読み取り値が閾値を超えたタイミング(例えば5分)にトリガーを設定することができます。

タグ付きのイベントタイムレポートを持つストリーミングフローのみが報告 stream_latency_msし、しかもSDPタイムメトリクスが有効である場合のみです。 他のフローは毎回 NULL 返ってきますが、このアラートはそれらに対しては決して発生しません。

-- Returns the most recent new-data latency per flow for a given pipeline.
-- The alert can trigger when stream_latency_ms > 300000 (5 minutes) for any flow.
WITH latest_latency AS (
  SELECT
    workspace_id,
    pipeline_id,
    origin.flow_name,
    event_time,
    CASE
      WHEN variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED' THEN 0
      ELSE variant_get(details, '$.flow_progress.streaming_metrics.stream_latency_ms', 'BIGINT')
    END AS stream_latency_ms
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    pipeline_id = '<your-pipeline-id>'
    AND event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
    AND (
      variant_get(details, '$.flow_progress.streaming_metrics.stream_latency_ms', 'BIGINT') IS NOT NULL
      OR variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED'
    )
  QUALIFY
    ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id, origin.flow_name ORDER BY event_time DESC) = 1
)
SELECT *
FROM latest_latency
WHERE stream_latency_ms > 300000

運用アラートのヒント

• 各アラートがアカウント全体ではなく特定のスコープを対象とするように、 pipeline_id (およびワークスペースごとにアラートを保持する場合は workspace_id ) でフィルター処理します。
• アラートの感度に一致する評価周期を選択します。 高速失敗信号には短い間隔 (5 分ごとなど) を、バックログとデータ品質の傾向には長い間隔 (時間単位など) を使用します。 "クエリが 0 行を超える行を返すときにトリガーする" 条件は、ほとんどの場合に機能します。

参照値

レベル値

満期レベルの価値

イベントタイプの値

event_type フィールドは列挙型です。 値の完全なセット: