Delta Live Tablesのイベントログ

[Delta Live Tables event log \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-event-log.html) [2022/1/13時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

:::note info
**プレビュー**
本機能は[パブリックプレビュー](https://docs.databricks.com/release-notes/release-types.html)です。サインアップには[Delta Live Tablesへのアクセスをリクエスト](https://databricks.com/p/product-delta-live-tables)してください。
:::

それぞれのDelta Live Tablesのパイプラインごとにイベントログが作成され、メンテナンスされます。イベントログには、監査ログ、データ品質チェック、パイプラインの進捗状況、データリネージュを含むパイプラインに関する全ての情報が含まれています。このイベントログを用いて、お使いのパイプラインの状態を追跡、理解、モニタリングすることができます。

それぞれのパイプラインに対するイベントログは、DBFS上のDeltaテーブルに格納されます。Delta Live Tablesの[ユーザーインタフェース](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-ui.html)、Delta Live Tables [API](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-api-guide.html#list-events)、あるいはDeltaテーブルを直接クエリーすることでイベントログのエントリーを参照することができます。

[サンプルノートブック](#サンプルノートブック)には、本書で論じられるクエリーが含まれており、Delta Live Tablesイベントログの探索に活用することができます。

# 要件

本文のサンプルでは、Databricksランタイム8.1以降で利用できる[JSON SQL](https://docs.databricks.com/spark/latest/spark-sql/semi-structured.html)関数を使用しています。

# イベントログの格納場所

イベントログは`storage`の配下の`/system/events`に格納されます。例えば、パイプラインの`storage`設定を`/Users/username/data`とした場合、イベントログはDBFSの`/Users/username/data/system/events`に格納されます。

`storage`を設定しない場合、デフォルトのイベントログの格納場所は、DBFSの`/pipelines/<pipeline-id>/system/events`となります。例えば、パイプラインのIDが`91de5e48-35ed-11ec-8d3d-0242ac130003`の場合、`/pipelines/91de5e48-35ed-11ec-8d3d-0242ac130003/system/events`に格納されます。

# イベントログのスキーマ

以下の表では、イベントログのスキーマを説明しています。これらのフィールドのいくつかは、クエリーを実行するためにパーシングが必要とあんるJSONドキュメントを含んでいます。例えば、データ品質メトリクスの分析には、`details` JSONドキュメントのフィールドのパーシングが必要となります。本書におけるサンプルでは、必要なパーシングを行うためにPython関数を用いてデモを行なっています。

| フィールド | 説明 |
|:--|:--|
| id  | パイプラインのID  |
| sequence  | イベントを識別し並び替えるためのメタデータを含むJSONドキュメント  |
| origin  | イベントの起源を示すメタデータを含むJSONドキュメント。例えば、クラウドプロバイダー、リージョン、ユーザーID、パイプラインID  |
|  timestamp | イベントが記録された時刻  |
| message  | イベントを説明する判読可能なメッセージ  |
| level  | イベントタイプ。例えば、INFO、WARN、ERROR、METRICS  |
| error  | エラーが発生した場合、エラーを説明する詳細メッセージ  |
| details  | イベントの詳細を含むJSONドキュメント。イベントを分析する際に用いられる主要なフィールド  |
| event_type  | イベントタイプ  |

# イベントログに対するクエリー

- [監査ログ](#監査ログ)
- [リネージュ](#リネージュ)
- [データ品質](#データ品質)

イベントログに対するクエリーをシンプルにするためにビューを作成することができます。以下の例では、`event_log_view`というビューを作成しています。以降の例では、イベントログのレコードにクエリーを実行するために、このビューを使用します。

```py:Python
event_log = spark.read.format('delta').load(event_log_path)
event_log.createOrReplaceTempView("event_log_raw")
```

`event_log_path`を[イベントログの格納場所](#イベントログの格納場所)で置き換えてください。

パイプライン実行のそれぞれのインスタンスは`update(アップデート)`と呼ばれます。以下のクエリーのいくつかにおいては、最新のアップデートに関する情報を抽出しています。最新のアップデートのIDを検索し、変数`latest_update_id`に保存するために、以下のコマンドを実行します。

```py:Python
latest_update_id = spark.sql("SELECT origin.update_id FROM event_log_raw WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1").collect()[0].update_id
spark.conf.set('latest_update.id', latest_update_id)
```

## 監査ログ

ユーザーのアクションのようなイベントを監査するためにイベントログを活用することができます。イベントには、`user_action`というイベントタイプを持つユーザーのアクションが含まれています。アクションに関する情報は`details`フィールドの`user_action`オブジェクトに格納されています。ユーザーイベントの監査ログを構成するために、以下のクエリーを使用します。

```sql:SQL
SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'
```
![Screen Shot 2022-03-10 at 9.39.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/44dfac99-a6bf-f886-1cc3-c98d7efd1923.png)

## リネージュ

Delta Live Tablesのユーザーインタフェースでパイプラインのグラフ構造のビジュアライゼーションを参照することができます。また、企業におけるコンプライアンスに対するレポートの作成やデータの依存関係の追跡を行うといったタスクを実行するために、この情報をプログラムから抽出することができます。リネージュに関する情報を含むイベントは、イベントタイプ`flow_definition`となっています。リネージュ情報は`details`フィールドの`flow_definition`に格納されています。`flow_definition`オブジェクトのフィールドには、データセット間の関係を推定するために必要な情報が含まれています。

```sql:SQL
SELECT details:flow_definition.output_dataset, details:flow_definition.input_datasets FROM event_log_raw WHERE event_type = 'flow_definition' AND origin.update_id = '${latest_update.id}'
```
![Screen Shot 2022-03-10 at 9.43.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bdcdee9e-6170-e72b-75ca-f9f3dfbb1611.png)

## データ品質

イベントログは、お使いのパイプラインで定義された[エクスペクテーション](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-expectations.html)に基づくデータ品質メトリクスも格納しています。データ品質に関する情報を含むイベントは、イベントタイプ`flow_progress`となります。データセットにエクスペクテーションが定義された際、データ品質メトリクスは`flow_progress.data_quality.expectations`オブジェクトの詳細フィールドに格納されます。以下のサンプルクエリーは、最新のパイプラインアップデートに対するデータ品質メトリクスを取得しています。

```sql:SQL
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
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = '${latest_update.id}'
  )
GROUP BY
  row_expectations.dataset,
  row_expectations.name
```

# サンプルノートブック

## Delta Live Tablesイベントログに対するクエリー

https://github.com/taka-yayoi/public_repo/tree/main/DLT_eventlog

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)