パイプライン イベント ログ

パイプライン イベント ログには、監査ログ、データ品質チェック、パイプラインの進行状況、データ系列など、パイプラインに関連するすべての情報が含まれます。 イベント ログを使用して、データ パイプラインの状態を追跡、把握、および監視することができます。

イベント ログ エントリは、パイプライン監視 ユーザー インターフェイス、 Pipelines REST API、またはイベント ログに直接クエリを実行して表示できます。 このセクションでは、イベント ログに直接クエリを実行することに重点を置いています。

また、イベント フックを使用してイベント (例: アラートの送信) がログに記録されると、実行するカスタム アクションを定義することもできます。

イベント ログ スキーマの詳細については、「 パイプライン イベント ログ スキーマ」を参照してください。

イベント ログのクエリを実行する

既定では、パイプラインは、パイプライン用に構成された既定のカタログとスキーマの非表示の Delta テーブルにイベント ログを書き込みます。 非表示の状態でも、十分な特権を持つすべてのユーザーがテーブルに対してクエリを実行できます。 既定では、パイプラインの 実行ユーザー のみがイベント ログ テーブルに対してクエリを実行できます。

実行ユーザーとしてイベント ログにクエリを実行するには、パイプライン ID を使用します。

SELECT * FROM event_log(<pipelineId>);

既定では、非表示のイベント ログの名前は event_log_{pipeline_id}形式になります。パイプライン ID はシステム割り当て UUID で、ダッシュはアンダースコアに置き換えられます。 イベント ログ テーブルは system.information_schema.tables に表示されますが、カタログ エクスプローラーやその他のワークスペース UI ページには表示されません。 event_log()関数を使用してアクセスする必要があります。

パイプラインの 詳細設定 を編集して、イベント ログを発行できます。 詳細については、 イベント ログのパイプライン設定を参照してください。 イベント ログを発行する場合は、イベント ログの名前を指定し、必要に応じて、次の例のようにカタログとスキーマを指定します。

{
  "id": "ec2a0ff4-d2a5-4c8c-bf1d-d9f12f10e749",
  "name": "billing_pipeline",
  "event_log": {
    "catalog": "catalog_name",
    "schema": "schema_name",
    "name": "event_log_table_name"
  }
}

イベント ログの場所は、パイプライン内の自動ローダー クエリのスキーマの場所としても機能します。 Databricks では、権限を変更する前にイベント ログ テーブルのビューを作成することをお勧めします。一部のコンピューティング設定では、イベント ログ テーブルが直接共有されている場合に、ユーザーがスキーマ メタデータにアクセスできる場合があるためです。 次の構文例では、イベント ログ テーブルにビューを作成し、この記事に含まれるイベント ログ クエリの例で使用します。 <catalog_name>.<schema_name>.<event_log_table_name>をパイプライン イベント ログの完全修飾テーブル名に置き換えます。 イベント ログを発行した場合は、発行時に指定した名前を使用します。 それ以外の場合は、 event_log(<pipelineId>) pipelineId がクエリするパイプラインの ID である場合に使用します。

CREATE VIEW event_log_raw
AS SELECT * FROM <catalog_name>.<schema_name>.<event_log_table_name>;

Unity カタログでは、ビューはストリーミング クエリをサポートします。 次の例では、構造化ストリーミングを使用して、イベント ログ テーブルの上に定義されたビューに対してクエリを実行します。

df = spark.readStream.table("event_log_raw")

基本的なクエリの例

次の例では、イベント ログに対してクエリを実行してパイプラインに関する一般的な情報を取得し、一般的なシナリオのデバッグに役立つ方法を示します。

以前の更新プログラムのクエリを実行してパイプラインの更新を監視する

次の例では、パイプラインの更新 (または実行) を照会し、更新 ID、状態、開始時刻、完了時間、期間を表示します。 これにより、パイプラインの実行の概要が示されます。

event_log_rawの説明に従って、目的のパイプラインの ビューを作成していることを前提としています。

with last_status_per_update AS (
    SELECT
        origin.pipeline_id AS pipeline_id,
        origin.pipeline_name AS pipeline_name,
        origin.update_id AS pipeline_update_id,
        FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state AS last_update_state,
        timestamp,
        ROW_NUMBER() OVER (
            PARTITION BY origin.update_id
            ORDER BY timestamp DESC
        ) AS rn
    FROM event_log_raw
    WHERE event_type = 'update_progress'
    QUALIFY rn = 1
),
update_durations AS (
    SELECT
        origin.pipeline_id AS pipeline_id,
        origin.pipeline_name AS pipeline_name,
        origin.update_id AS pipeline_update_id,
        -- Capture the start of the update
        MIN(CASE WHEN event_type = 'create_update' THEN timestamp END) AS start_time,

        -- Capture the end of the update based on terminal states or current timestamp (relevant for continuous mode pipelines)
        COALESCE(
            MAX(CASE
                WHEN event_type = 'update_progress'
                 AND FROM_JSON(details, 'struct<update_progress: struct<state: string>>').update_progress.state IN ('COMPLETED', 'FAILED', 'CANCELED')
                THEN timestamp
            END),
            current_timestamp()
        ) AS end_time
    FROM event_log_raw
    WHERE event_type IN ('create_update', 'update_progress')
      AND origin.update_id IS NOT NULL
    GROUP BY pipeline_id, pipeline_name, pipeline_update_id
    HAVING start_time IS NOT NULL
)
SELECT
    s.pipeline_id,
    s.pipeline_name,
    s.pipeline_update_id,
    d.start_time,
    d.end_time,
    CASE
        WHEN d.start_time IS NOT NULL AND d.end_time IS NOT NULL THEN
            ROUND(TIMESTAMPDIFF(MILLISECOND, d.start_time, d.end_time) / 1000)
        ELSE NULL
    END AS duration_seconds,
    s.last_update_state AS pipeline_update_status
FROM last_status_per_update s
JOIN update_durations d
  ON s.pipeline_id = d.pipeline_id
 AND s.pipeline_update_id = d.pipeline_update_id
ORDER BY d.start_time DESC;

具体化されたビューの増分更新の問題をデバッグする

この例では、パイプラインの最新の更新からのすべてのフローを照会します。 増分更新されたかどうか、および増分更新が発生しない理由のデバッグに役立つその他の関連する計画情報が表示されます。

event_log_rawの説明に従って、目的のパイプラインの ビューを作成していることを前提としています。

WITH latest_update AS (
  SELECT
    origin.pipeline_id,
    origin.update_id AS latest_update_id
  FROM event_log_raw AS origin
  WHERE origin.event_type = 'create_update'
  ORDER BY timestamp DESC
  -- LIMIT 1 -- remove if you want to get all of the update_ids
),
parsed_planning AS (
  SELECT
    origin.pipeline_name,
    origin.pipeline_id,
    origin.flow_name,
    lu.latest_update_id,
    from_json(
      details:planning_information,
      'struct<
        technique_information: array<struct<
          maintenance_type: string,
          is_chosen: boolean,
          is_applicable: boolean,
          cost: double,
          incrementalization_issues: array<struct<
            issue_type: string,
            prevent_incrementalization: boolean,
            operator_name: string,
            plan_not_incrementalizable_sub_type: string,
            expression_name: string,
            plan_not_deterministic_sub_type: string
          >>
        >>
      >'
    ) AS parsed
  FROM event_log_raw AS origin
  JOIN latest_update lu
    ON origin.update_id = lu.latest_update_id
  WHERE details:planning_information IS NOT NULL
),
chosen_technique AS (
  SELECT
    pipeline_name,
    pipeline_id,
    flow_name,
    latest_update_id,
    FILTER(parsed.technique_information, t -> t.is_chosen = true)[0] AS chosen_technique,
    parsed.technique_information AS planning_information
  FROM parsed_planning
)
SELECT
  pipeline_name,
  pipeline_id,
  flow_name,
  latest_update_id,
  chosen_technique.maintenance_type,
  chosen_technique,
  planning_information
FROM chosen_technique
ORDER BY latest_update_id DESC;

パイプライン更新のコストを照会する

この例では、パイプラインの DBU 使用量と、特定のパイプライン実行のユーザーに対してクエリを実行する方法を示します。

SELECT
  sku_name,
  billing_origin_product,
  usage_date,
  collect_set(identity_metadata.run_as) as users,
  SUM(usage_quantity) AS `DBUs`
FROM
  system.billing.usage
WHERE
  usage_metadata.dlt_pipeline_id = :pipeline_id
GROUP BY
  ALL;

詳細なクエリ

次の例では、イベント ログに対してクエリを実行して、あまり一般的ではない、または高度なシナリオを処理する方法を示します。

パイプライン内のすべてのフローのクエリ メトリック

この例では、パイプライン内のすべてのフローに関する詳細情報を照会する方法を示します。 フロー名、更新期間、データ品質メトリック、処理された行 (出力行、削除、アップサート、および削除されたレコード) に関する情報が表示されます。

event_log_rawの説明に従って、目的のパイプラインの ビューを作成していることを前提としています。

WITH flow_progress_raw AS (
  SELECT
    origin.pipeline_name         AS pipeline_name,
    origin.pipeline_id           AS pipeline_id,
    origin.flow_name             AS table_name,
    origin.update_id             AS update_id,
    timestamp,
    details:flow_progress.status AS status,
    TRY_CAST(details:flow_progress.metrics.num_output_rows AS BIGINT)      AS num_output_rows,
    TRY_CAST(details:flow_progress.metrics.num_upserted_rows AS BIGINT)    AS num_upserted_rows,
    TRY_CAST(details:flow_progress.metrics.num_deleted_rows AS BIGINT)     AS num_deleted_rows,
    TRY_CAST(details:flow_progress.data_quality.dropped_records AS BIGINT) AS num_expectation_dropped_rows,
    FROM_JSON(
      details:flow_progress.data_quality.expectations,
      SCHEMA_OF_JSON("[{'name':'str', 'dataset':'str', 'passed_records':42, 'failed_records':42}]")
    ) AS expectations_array

  FROM event_log_raw
  WHERE event_type = 'flow_progress'
    AND origin.flow_name IS NOT NULL
    AND origin.flow_name != 'pipelines.flowTimeMetrics.missingFlowName'
),

aggregated_flows AS (
  SELECT
    pipeline_name,
    pipeline_id,
    update_id,
    table_name,
    MIN(CASE WHEN status IN ('STARTING', 'RUNNING', 'COMPLETED') THEN timestamp END) AS start_timestamp,
    MAX(CASE WHEN status IN ('STARTING', 'RUNNING', 'COMPLETED') THEN timestamp END) AS end_timestamp,
    MAX_BY(status, timestamp) FILTER (
      WHERE status IN ('COMPLETED', 'FAILED', 'CANCELLED', 'EXCLUDED', 'SKIPPED', 'STOPPED', 'IDLE')
    ) AS final_status,
    SUM(COALESCE(num_output_rows, 0))              AS total_output_records,
    SUM(COALESCE(num_upserted_rows, 0))            AS total_upserted_records,
    SUM(COALESCE(num_deleted_rows, 0))             AS total_deleted_records,
    MAX(COALESCE(num_expectation_dropped_rows, 0)) AS total_expectation_dropped_records,
    MAX(expectations_array)                        AS total_expectations

  FROM flow_progress_raw
  GROUP BY pipeline_name, pipeline_id, update_id, table_name
)
SELECT
  af.pipeline_name,
  af.pipeline_id,
  af.update_id,
  af.table_name,
  af.start_timestamp,
  af.end_timestamp,
  af.final_status,
  CASE
    WHEN af.start_timestamp IS NOT NULL AND af.end_timestamp IS NOT NULL THEN
      ROUND(TIMESTAMPDIFF(MILLISECOND, af.start_timestamp, af.end_timestamp) / 1000)
    ELSE NULL
  END AS duration_seconds,

  af.total_output_records,
  af.total_upserted_records,
  af.total_deleted_records,
  af.total_expectation_dropped_records,
  af.total_expectations
FROM aggregated_flows af
-- Optional: filter to latest update only
WHERE af.update_id = (
  SELECT update_id
  FROM aggregated_flows
  ORDER BY end_timestamp DESC
  LIMIT 1
)
ORDER BY af.end_timestamp DESC, af.pipeline_name, af.pipeline_id, af.update_id, af.table_name;

データ品質または期待値のメトリックに対してクエリを実行する

パイプライン内のデータ セットに対する期待値を定義すると、予測に合格および失敗したレコードの数のメトリックが details:flow_progress.data_quality.expectations オブジェクトに格納されます。 削除されたレコードの数のメトリックは、 details:flow_progress.data_quality オブジェクトに格納されます。 データ品質に関する情報を含むイベントには、イベントの種類 flow_progress があります。

一部のデータ セットでは、データ品質メトリックを使用できない場合があります。 期待される 制限事項を参照してください。

次のデータ品質メトリックを使用できます。

次の例では、最後のパイプライン更新のデータ品質メトリックを照会します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

WITH latest_update AS (
  SELECT
    origin.pipeline_id,
    origin.update_id AS latest_update_id
  FROM event_log_raw AS origin
  WHERE origin.event_type = 'create_update'
  ORDER BY timestamp DESC
  LIMIT 1 -- remove if you want to get all of the update_ids
)
SELECT
  row_expectations.dataset as dataset,
  row_expectations.name as expectation,
  SUM(row_expectations.passed_records) as passing_records,
  SUM(row_expectations.failed_records) as failing_records
FROM
  (
    SELECT
      explode(
        from_json(
          details:flow_progress:data_quality:expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations
    FROM
      event_log_raw,
      latest_update
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = latest_update.latest_update_id
  )
GROUP BY
  row_expectations.dataset,
  row_expectations.name;

系列情報のクエリ

データ系列に関する情報を含むイベントには、イベントの種類 flow_definition があります。 details:flow_definition オブジェクトには、グラフ内の各リレーションシップを定義する output_dataset と input_datasets が含まれます。

次のクエリを使用して、入力データ セットと出力データ セットを抽出して系列情報を表示します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  details:flow_definition.output_dataset as flow_name,
  details:flow_definition.input_datasets as input_flow_names,
  details:flow_definition.flow_type as flow_type,
  details:flow_definition.schema, -- the schema of the flow
  details:flow_definition -- overall flow_definition object
FROM event_log_raw inner join latest_update on origin.update_id = latest_update.id
WHERE details:flow_definition IS NOT NULL
ORDER BY timestamp;

自動ローダーを使用してクラウド ファイルインジェストを監視する

パイプラインは、自動ローダーがファイルを処理するときにイベントを生成します。 自動ローダー イベントの場合、event_type は operation_progress され、details:operation_progress:type は AUTO_LOADER_LISTING または AUTO_LOADER_BACKFILLです。 details:operation_progress オブジェクトには、status、duration_ms、auto_loader_details:source_path、および auto_loader_details:num_files_listed フィールドも含まれます。

次の例では、自動ローダー イベントに対して最新の更新プログラムのクエリを実行します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  timestamp,
  details:operation_progress.status,
  details:operation_progress.type,
  details:operation_progress:auto_loader_details
FROM
  event_log_raw,latest_update
WHERE
  event_type like 'operation_progress'
  AND
  origin.update_id = latest_update.id
  AND
  details:operation_progress.type in ('AUTO_LOADER_LISTING', 'AUTO_LOADER_BACKFILL');

ストリーミング期間を最適化するためにデータ バックログを監視する

各パイプラインは、 details:flow_progress.metrics.backlog_bytes オブジェクトのバックログに存在するデータの量を追跡します。 バックログ メトリックを含むイベントには、イベントの種類 flow_progress があります。 次の例では、最後のパイプライン更新のバックログ メトリックを照会します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  timestamp,
  Double(details :flow_progress.metrics.backlog_bytes) as backlog
FROM
  event_log_raw,
  latest_update
WHERE
  event_type ='flow_progress'
  AND
  origin.update_id = latest_update.id;

自動スケール イベントを監視してクラシック コンピューティングを最適化する

クラシック コンピューティングを使用するパイプライン (つまり、サーバーレス コンピューティングを使用しない) の場合、イベント ログは、パイプラインで拡張自動スケーリングが有効になっている場合にクラスターのサイズ変更をキャプチャします。 拡張自動スケールに関する情報を含むイベントには、イベントの種類が autoscale。 クラスターのサイズ変更要求の情報は、details:autoscale オブジェクトに格納されます。

次の例では、最後のパイプライン更新に対する拡張自動スケール クラスターのサイズ変更要求を照会します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  timestamp,
  Double(
    case
      when details :autoscale.status = 'RESIZING' then details :autoscale.requested_num_executors
      else null
    end
  ) as starting_num_executors,
  Double(
    case
      when details :autoscale.status = 'SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'PARTIALLY_SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as partially_succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'FAILED' then details :autoscale.requested_num_executors
      else null
    end
  ) as failed_num_executors
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'autoscale'
  AND
  origin.update_id = latest_update.id

クラシック コンピューティングのコンピューティング リソース使用率を監視する

cluster_resources イベントは、クラスター内のタスク スロットの数、それらのタスク スロットの使用量、スケジュールを待機しているタスクの数に関するメトリックを提供します。

拡張自動スケールが有効になっている場合、 cluster_resources イベントには、 latest_requested_num_executorsや optimal_num_executorsなど、自動スケール アルゴリズムのメトリックも含まれます。 イベントには、アルゴリズムの状態も、CLUSTER_AT_DESIRED_SIZE、SCALE_UP_IN_PROGRESS_WAITING_FOR_EXECUTORS、BLOCKED_FROM_SCALING_DOWN_BY_CONFIGURATION などのさまざまな状態として表示されます。 この情報は、自動スケール イベントと組み合わせて表示して、拡張自動スケールの全体像を提供できます。

次の例では、タスク キューのサイズ履歴、使用率履歴、Executor カウント履歴、およびその他のメトリックと状態をクエリして、最後のパイプライン更新での自動スケールを行います。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  timestamp,
  Double(details:cluster_resources.avg_num_queued_tasks) as queue_size,
  Double(details:cluster_resources.avg_task_slot_utilization) as utilization,
  Double(details:cluster_resources.num_executors) as current_executors,
  Double(details:cluster_resources.latest_requested_num_executors) as latest_requested_num_executors,
  Double(details:cluster_resources.optimal_num_executors) as optimal_num_executors,
  details :cluster_resources.state as autoscaling_state
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id;

パイプライン ストリーミング メトリックを監視する

パイプラインでストリームの進行状況に関するメトリックを表示できます。 次 の例外 を除き、構造化ストリーミングによって作成された StreamingQueryListener メトリック とよく似たイベントを取得するために、stream_progress イベントのクエリを実行します。

• StreamingQueryListenerには次のメトリックが存在しますが、stream_progressには存在しません:numInputRows、inputRowsPerSecond、およびprocessedRowsPerSecond。
• Kafka および Kineses ストリームの場合、 startOffset、 endOffset、および latestOffset フィールドが大きすぎる可能性があり、切り捨てられます。 これらの各フィールドに対して、データが切り捨てられているかどうかを示すブール値を使用して、追加の ...Truncated フィールド、 startOffsetTruncated、 endOffsetTruncated、および latestOffsetTruncatedが追加されます。

stream_progress イベントのクエリを実行するには、次のようなクエリを使用します。

SELECT
  parse_json(get_json_object(details, '$.stream_progress.progress_json')) AS stream_progress_json
FROM event_log_raw
WHERE event_type = 'stream_progress';

JSON でのイベントの例を次に示します。

{
  "id": "abcd1234-ef56-7890-abcd-ef1234abcd56",
  "sequence": {
    "control_plane_seq_no": 1234567890123456
  },
  "origin": {
    "cloud": "<cloud>",
    "region": "<region>",
    "org_id": 123456789012345,
    "pipeline_id": "abcdef12-abcd-3456-7890-abcd1234ef56",
    "pipeline_type": "WORKSPACE",
    "pipeline_name": "<pipeline name>",
    "update_id": "1234abcd-ef56-7890-abcd-ef1234abcd56",
    "request_id": "1234abcd-ef56-7890-abcd-ef1234abcd56"
  },
  "timestamp": "2025-06-17T03:18:14.018Z",
  "message": "Completed a streaming update of 'flow_name'.",
  "level": "INFO",
  "details": {
    "stream_progress": {
      "progress": {
        "id": "abcdef12-abcd-3456-7890-abcd1234ef56",
        "runId": "1234abcd-ef56-7890-abcd-ef1234abcd56",
        "name": "silverTransformFromBronze",
        "timestamp": "2022-11-01T18:21:29.500Z",
        "batchId": 4,
        "durationMs": {
          "latestOffset": 62,
          "triggerExecution": 62
        },
        "stateOperators": [],
        "sources": [
          {
            "description": "DeltaSource[dbfs:/path/to/table]",
            "startOffset": {
              "sourceVersion": 1,
              "reservoirId": "abcdef12-abcd-3456-7890-abcd1234ef56",
              "reservoirVersion": 3216,
              "index": 3214,
              "isStartingVersion": true
            },
            "endOffset": {
              "sourceVersion": 1,
              "reservoirId": "abcdef12-abcd-3456-7890-abcd1234ef56",
              "reservoirVersion": 3216,
              "index": 3214,
              "isStartingVersion": true
            },
            "latestOffset": null,
            "metrics": {
              "numBytesOutstanding": "0",
              "numFilesOutstanding": "0"
            }
          }
        ],
        "sink": {
          "description": "DeltaSink[dbfs:/path/to/sink]",
          "numOutputRows": -1
        }
      }
    }
  },
  "event_type": "stream_progress",
  "maturity_level": "EVOLVING"
}

次の例では、Kafka ソース内の非切り捨てレコードが示されており、...Truncated フィールドが false に設定されています。

{
  "description": "KafkaV2[Subscribe[KAFKA_TOPIC_NAME_INPUT_A]]",
  "startOffsetTruncated": false,
  "startOffset": {
    "KAFKA_TOPIC_NAME_INPUT_A": {
      "0": 349706380
    }
  },
  "endOffsetTruncated": false,
  "endOffset": {
    "KAFKA_TOPIC_NAME_INPUT_A": {
      "0": 349706672
    }
  },
  "latestOffsetTruncated": false,
  "latestOffset": {
    "KAFKA_TOPIC_NAME_INPUT_A": {
      "0": 349706672
    }
  },
  "numInputRows": 292,
  "inputRowsPerSecond": 13.65826278123392,
  "processedRowsPerSecond": 14.479817514628582,
  "metrics": {
    "avgOffsetsBehindLatest": "0.0",
    "estimatedTotalBytesBehindLatest": "0.0",
    "maxOffsetsBehindLatest": "0",
    "minOffsetsBehindLatest": "0"
  }
}

監査パイプライン

イベント ログ レコードやその他の Azure Databricks 監査ログを使用して、パイプラインでデータがどのように更新されているかを把握できます。

パイプラインは所有者の認証情報で更新を行います。 パイプライン所有者を変更することで、使用する資格情報 を変更できます。 監査ログには、パイプラインの作成、構成の編集、更新のトリガーなど、パイプラインでのアクションのユーザーが記録されます。

Unity カタログ監査イベントのリファレンスについては、Unity カタログ イベント を参照してください。

イベント ログにユーザー アクションのクエリを実行する

イベント ログを使用して、ユーザー アクションなどのイベントを監査できます。 ユーザー アクションに関する情報を含むイベントには、イベントの種類 user_action があります。

アクションに関する情報は、user_action フィールドの details オブジェクトに格納されます。 次のクエリを使用して、ユーザー イベントの監査ログを作成します。 これは、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューを作成していることを前提としています。

SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'

ランタイム情報

パイプライン更新プログラムのランタイム情報 (更新プログラムの Databricks Runtime バージョンなど) を表示できます。この例では、「event_log_raw」の説明に従って、関心のあるパイプラインの ビューが作成されていることを前提としています。

SELECT origin.update_id, details:runtime_details:runtime_version:dbr_version FROM event_log_raw WHERE event_type = 'runtime_details'