Verkadaからデータを取り込む

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

このページでは、Lakeflow Connectを使用してマネージドVerkadaインジェストパイプラインを作成する方法について説明します。

要件

• 取り込みパイプラインを作成するには、まず次の要件を満たす必要があります：

  • ワークスペースでUnity Catalogが有効になっている必要があります。

  • ワークスペースでServerlessコンピュートを有効にする必要があります。Serverlessコンピュートの要件を参照してください。

  • 新しい接続を作成するには、メタストアに対する CREATE CONNECTION 権限が必要です。「Unity Catalog での権限の管理」を参照してください。

    コネクタが UI ベースのパイプライン オーサリングをサポートしている場合、管理者はこのページのステップを完了することで、接続とパイプラインを同時に作成できます。 ただし、パイプラインを作成するユーザーが API ベースのパイプライン オーサリングを使用している場合、または管理者以外のユーザーである場合、管理者はまずカタログ エクスプローラーで接続を作成する必要があります。 「管理対象取り込みソースへの接続」を参照してください。

  • 既存の接続を使用するには、接続オブジェクトに対するUSE CONNECTION権限またはALL PRIVILEGES権限が必要です。

  • ターゲットカタログに対する USE CATALOG 権限が必要です。

  • 既存のスキーマに対する USE SCHEMA 権限および CREATE TABLE 権限、またはターゲットカタログに対する CREATE SCHEMA 権限が必要です。

• Verkada から取り込むには、まず Databricks から認証を構成し、接続を作成します。「Verkada への認証の構成」および「Verkada 接続の作成」を参照してください。

コネクター オプション

個々のオブジェクトでテーブルスコープのオプションを設定します。使用方法については、例を参照してください。

オプション	スコープ	必須	適用対象	説明
start_datetime	テーブル	No	audit_log	初回同期時に取り込む最も古い監査イベント（ISO 8601 UTCのTimestamp形式、例：2026-07-09T12:00:00Z）。default to 365 days before the first sync, and no earlier than 2025-08-12.

オプション	スコープ	必須	適用対象	説明
start_datetime	テーブル	No	audit_log	初回同期時に取り込む最も古い監査イベント（ISO 8601 UTCのTimestamp形式、例：2026-07-09T12:00:00Z）。default to 365 days before the first sync, and no earlier than 2025-08-12.

取り込みパイプラインを作成

サポートされているソーステーブルの一覧については、「サポートされているソーステーブル」を参照してください。

Databricks UI

1. Databricksワークスペースのサイドバーで、 データ取り込み をクリックします。
2. データの追加 ページの Databricks コネクタ で、 Verkada をクリックします。
3. 取り込みウィザードの 接続 ページで、Verkadaの認証情報を保存している接続を選択します。メタストアに対するCREATE CONNECTION特権がある場合は、 接続の作成 をクリックして、Verkadaの認証を構成するの認証情報を使用して接続を作成します。
4. 次へ をクリックします。
5. インジェスト設定 ページで、パイプラインの名前を入力します。
6. イベント Logs の書き込み先となるカタログとスキーマを選択します。カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限がある場合は、ドロップダウンメニューの スキーマを作成 をクリックしてスキーマを作成します。
7. パイプラインの作成および続行 をクリックします。
8. ソース ページで、取り込むテーブルを選択します。
9. 保存して続行 をクリックします。
10. 宛先 ページで、データを読み込むカタログとスキーマを選択します。カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限がある場合は、ドロップダウンメニューの スキーマを作成 をクリックしてスキーマを作成します。
11. 保存して続行 をクリックします。
12. (オプション) スケジュールと通知 ページで、 スケジュールの作成 をクリックします。宛先テーブルを更新する頻度を設定します。
13. (オプション) 通知の追加 をクリックして、パイプライン操作の成功または失敗に関するEメール通知を設定し、 保存してパイプラインを実行 をクリックします。

Declarative Automation Bundles

宣言型オートメーションバンドルを使用して、Verkada パイプラインをコードとして管理します。バンドルにはジョブとタスクの YAML 定義を含めることができ、Databricks CLI を使用して管理され、異なるターゲットワークスペース（開発、ステージング、本番運用など）で共有および実行できます。詳細については、「宣言型オートメーションバンドルとは？」をご覧ください。

1. Databricks CLIを使用してバンドルを作成するには、次の手順を実行します。

   Bash

   databricks bundle init

2. バンドルに2つの新しいリソースファイルを追加します：

   • パイプライン定義ファイル（例：resources/verkada_pipeline.yml）。「pipeline.ingestion_definition」を参照してください。および「例」を参照してください。
   • データ取り込みの頻度を制御するジョブ定義ファイル（例：resources/verkada_job.yml）。

3. Databricks CLIを使用してパイプラインをデプロイします：

   Bash

   databricks bundle deploy

Databricks notebook

1. 次のノートブックを Databricks ワークスペースにインポートします：

ノートブックを新しいタブで開くOpen in Databricks

2. セル1とセル2はそのままにしておきます。変更しないでください。

3. セル3を、パイプラインの構成詳細で変更します。「パイプライン.ingestion_definition」を参照してください。および「例」。

4. 必要に応じて、高度なパイプライン設定を構成します。「管理された取り込みパイプラインの一般的なパターン」を参照してください。

5. 「 すべて実行 」をクリックします。

例

Verkadaコネクタは、defaultソーススキーマ内の2つのソーステーブルを利用可能にします。個々のテーブルまたはスキーマ全体を取り込みます。

特定のテーブルを取り込む

このオプションを使用して、テーブルの特定のサブセットを取り込むか、テーブルごとに宛先の命名をカスタマイズします。

Declarative Automation Bundles

次のパイプライン定義ファイルは、個々のVerkadaテーブルを取り込みます：

YAML

resources:
  pipelines:
    verkada_pipeline:
      name: verkada_pipeline
      catalog: 'main'
      target: 'verkada_data'
      ingestion_definition:
        connection_name: verkada_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'audit_log'
              destination_catalog: 'main'
              destination_schema: 'verkada_data'
              destination_table: 'audit_log'
              connector_options:
                api_source_connector_options:
                  options:
                    start_datetime: '<start-datetime>'
          - table:
              source_schema: 'default'
              source_table: 'users'
              destination_catalog: 'main'
              destination_schema: 'verkada_data'
              destination_table: 'users'

Databricks notebook

次のパイプライン仕様は、個々の Verkada テーブルを取り込みます：

Python

pipeline_name = "verkada_pipeline"
connection_name = "<verkada-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "audit_log",
          "destination_catalog": "main",
          "destination_schema": "verkada_data",
          "destination_table": "audit_log",
          "connector_options": {
            "api_source_connector_options": {
              "options": {
                "start_datetime": "<start-datetime>"
              }
            }
          }
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "users",
          "destination_catalog": "main",
          "destination_schema": "verkada_data",
          "destination_table": "users"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

スキーマ全体を取り込む

このオプションを使用して、1つの宣言でVerkadaのすべてのソーステーブルを単一の宛先スキーマに取り込みます。

Declarative Automation Bundles

次のパイプライン定義ファイルは、サポートされているすべての Verkada テーブルを宛先スキーマに取り込みます：

YAML

resources:
  pipelines:
    verkada_pipeline:
      name: verkada_pipeline
      catalog: 'main'
      target: 'verkada_data'
      ingestion_definition:
        connection_name: verkada_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'verkada_data'

Databricks notebook

次のパイプライン仕様では、サポートされているすべてのVerkadaテーブルを宛先スキーマに取り込みます：

Python

pipeline_name = "verkada_pipeline"
connection_name = "<verkada-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "schema": {
          "source_schema": "default",
          "destination_catalog": "main",
          "destination_schema": "verkada_data"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

宣言型オートメーションバンドルのジョブ定義ファイル

以下は、宣言型オートメーションバンドルで使用するジョブ定義ファイルの例です。ジョブは毎日実行されます。

Declarative Automation Bundles

YAML

resources:
  jobs:
    verkada_job:
      name: verkada_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: verkada_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.verkada_pipeline.id}

一般的なパターン

高度なパイプライン構成については、「管理されたインジェストパイプラインの一般的なパターン」を参照してください。

次のステップ

パイプラインの起動、スケジュール、アラートを設定します。一般的なパイプラインのメンテナンス タスクを参照してください。

その他のリソース

• VerkadaコネクタのFAQ
• Verkadaコネクタの制限事項
• Verkada コネクタ リファレンス
• Verkadaコネクターのトラブルシューティング