概要

Observability Pipelines の Databricks (Zerobus) 宛先を使用して、ログを Databricks Unity Catalog テーブルに送信します。送信先はログを Zerobus Ingest API にストリーミングし、OAuth サービス プリンシパルを使用して Databricks に認証します。

前提条件

Databricks (Zerobus) 宛先を構成する前に、次のことを行う必要があります。

• Observability Pipelines Worker がログを書き込む Unity Catalog スキーマとテーブルのセットアップ。
• Worker が Databricks に認証するために使用するサービス プリンシパルのセットアップ。このサービス プリンシパルには、テーブルの読み取りおよび書き込みの権限が必要です。

スキーマとテーブルのセットアップ

このセクションの SQL 例では、次のプレースホルダーを使用します。

注意: GRANT コマンドは Databricks ワークスペース管理者によって実行される必要があります。

Databricks ワークスペース内で以下を行います。

1. Databricks ワークスペース管理者でない場合は、管理者に次のコマンドを実行して、スキーマを作成する権限をユーザーに付与してもらってください。

   GRANT CREATE SCHEMA ON CATALOG <CATALOG_NAME> TO <USER>;
   

2. スキーマを作成します。

   CREATE SCHEMA IF NOT EXISTS <CATALOG_NAME>.<SCHEMA_NAME>
   MANAGED LOCATION '<YOUR_MANAGED_LOCATION>';
   

   • Note: MANAGED LOCATION is optional. See Databricks’ Create Schemas documentation for more information.

3. 管理者ユーザーでない場合は、管理者に次のコマンドを実行して、スキーマでテーブルを作成する権限をユーザーに付与してもらってください。

   GRANT CREATE TABLE ON SCHEMA <CATALOG_NAME>.<SCHEMA_NAME> TO <USER>;
   

4. Observability Pipelines がログデータを書き込むテーブルを作成するために、次のコマンドを実行します。

   CREATE TABLE <CATALOG_NAME>.<SCHEMA_NAME>.<TABLE_NAME> (
     host STRING,
     message STRING,
     service STRING,
     source_type STRING,
     timestamp TIMESTAMP
   );
   

   • See Databricks’ Create a Unity Catalog Managed Table documentation for more information.

完全修飾テーブル名は catalog.schema.table で、例えば main.obs_pipelines.apache_common_logs です。これは、Observability Pipelines Databricksの宛先をセットアップする際にテーブル名に入力する値です。

サービスプリンシパルのセットアップ

Databricks の Zerobus Ingest API は OAuth 認証を使用します。サービスプリンシパルを作成すると、OAuth クライアントシークレットが生成され、OAuth クライアント ID はサービスプリンシパルの UUID になります。

サービスプリンシパルを作成するには、以下を行います。

1. Databricks ワークスペースで、[User Settings] (ユーザー設定) > [Identity and access] (アイデンティティとアクセス) > [Service principals] (サービスプリンシパル) に移動します。
2. [Add service principal] (サービスプリンシパルを追加) をクリックします。
3. サービスプリンシパルが作成された後、そのための OAuth シークレットを生成します。
   • サービスプリンシパルのアプリケーション ID (クライアント ID) と OAuth クライアントシークレットをメモしておいてください。Observability Pipelines Databricks の宛先を構成する際に、両方が必要です。
4. Databricks でこの SQL を実行して、サービスプリンシパルにカタログ、スキーマ、およびテーブルへのアクセスを付与します。<SERVICE_PRINCIPAL_UUID> を前のステップでメモしたサービスプリンシパルのアプリケーション ID で置き換えます。

   GRANT USE CATALOG ON CATALOG <CATALOG_NAME> TO <SERVICE_PRINCIPAL_UUID>;
   GRANT USE SCHEMA ON SCHEMA <CATALOG_NAME>.<SCHEMA_NAME> TO <SERVICE_PRINCIPAL_UUID>;
   GRANT SELECT, MODIFY ON TABLE <CATALOG_NAME>.<SCHEMA_NAME>.<TABLE_NAME> TO <SERVICE_PRINCIPAL_UUID>;
   

詳細については、Databricks のアカウントにサービスプリンシパルを追加するおよびオブジェクトに対する権限を付与するのドキュメントを参照してください。

セットアップ

パイプラインをセットアップするときに、Databricks (Zerobus) 宛先を構成します。パイプラインのセットアップは、UI で、API を使用して、または Terraform で行えます。このセクションの手順は UI で構成されます。

注意: テーブルスキーマに存在しないログフィールドはドロップされます。例えば、ログにフィールド id、name、host があるのに、テーブルスキーマに name と host の列しか含まれていない場合、id フィールドはドロップされ、テーブルには書き込まれません。

パイプライン UI で Databricks (Zerobus) 宛先を選択した後:

1. Databricks ワークスペースの取り込みエンドポイントを入力します。https://<workspace_id>.zerobus.<region>.cloud.databricks.com などです。ワーカーはこのエンドポイントにログを送信します。
2. テーブル名 を catalog.schema.table の形式で入力します。main.obs_pipelines.apache_common_logs などです。
3. Databricksワークスペース用の Unity Catalog エンドポイントを入力します。https://<workspace>.cloud.databricks.com などです。ワーカーはこのエンドポイントを使用してテーブルのスキーマを読み取ります。
4. 認証 - クライアントID フィールドに、サービスプリンシパルのアプリケーションIDを入力します。abcdefgh-1234-5678-abcd-ef0123456789 などです。
5. Auth - Client Secret フィールドに、OAuth クライアントシークレットの識別子を入力します。空白のままにすると、デフォルトが使用されます。

オプション設定

バッファリング

Toggle the switch to enable Buffering Options. Enable a configurable buffer on your destination to ensure intermittent latency or an outage at the destination doesn’t create immediate backpressure, and allow events to continue to be ingested from your source. Disk buffers can also increase pipeline durability by writing data to disk, ensuring buffered data persists through a Worker restart. See Destination buffers for more information.

• If left unconfigured, your destination uses a memory buffer with a capacity of 500 events.
• To configure a buffer on your destination:
  1. Select the buffer type you want to set (Memory or Disk).
  2. Enter the buffer size and select the unit.
     1. Maximum memory buffer size is 128 GB.
     2. Maximum disk buffer size is 500 GB.
  3. In the Behavior on full buffer dropdown menu, select whether you want to block events or drop new events when the buffer is full.

文字列タイムスタンプをタイムスタンプ形式に変換する

ログに文字列形式のタイムスタンプがあり、Databricks テーブルにTIMESTAMP タイプとして宣言されたタイムスタンプ列がある場合、ログを Databricks (Zerobus) 宛先に送信する前に、文字列をタイムスタンプ形式に変換する必要があります。Databricks (Zerobus) は、タイムスタンプ形式しかその TIMESTAMP 型に変換できません。

文字列のタイムスタンプを変換しないと、ワーカーは次のようなエラーをスローします。

Protobuf encoding failed: Error converting timestamp field: Can't convert '2012-04-23T10[41]15Z' to i64: invalid digit found in string

文字列形式のタイムスタンプをタイムスタンプ形式に変換するには:

1. パイプラインにカスタムプロセッサを追加します。
2. 次のカスタムスクリプトを持つ関数を追加します。

   .timestamp = parse_timestamp!(.timestamp, format: "%+")
   

   See parse_timestamp for more information.

シークレットのデフォルト

These are the defaults used for secret identifiers and environment variables.

宛先の動作方法

イベントのバッチ処理

これらのパラメータのいずれかが満たされると、イベントのバッチがフラッシュされます。詳細についてはイベントのバッチ処理を参照してください。