メインコンテンツまでスキップ
に最終更新

Amplitude からデータを取り込む

備考

ベータ版

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

このページでは、Lakeflow Connect を使用して管理対象の Amplitude 取り込みパイプラインを作成する方法を示します。

要件​

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

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

    • ワークスペースで Serverless コンピュートを有効にする必要があります。See Serverless コンピュートの要件.

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

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

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

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

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

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

コネクターのオプション​

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

オプション

スコープ

必須

適用対象

説明

start_datetime

テーブル

No

events

最初の同期時に取り込む最も古いイベントのTimestamp（yyyy-MM-dd'T'HH:mm:ss'Z' 形式）。default は 1970-01-01T00:00:00Z です。

request_time_range

テーブル

No

events

各エクスポートウィンドウのサイズ（ISO-8601期間形式）。default は PT24H です。最小値は PT2H です。

オプション

スコープ

必須

適用対象

説明

start_datetime

テーブル

No

events

最初の同期時に取り込む最も古いイベントのTimestamp（yyyy-MM-dd'T'HH:mm:ss'Z' 形式）。default は 1970-01-01T00:00:00Z です。

request_time_range

テーブル

No

events

各エクスポートウィンドウのサイズ（ISO-8601期間形式）。default は PT24H です。最小値は PT2H です。

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

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

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

例​

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

注記

Amplitude コネクタがベータ版である間は、例に示すように、パイプラインを PREVIEW チャンネルにピン留めしてください。コネクタが一般提供されるまで、channel フィールドを PREVIEW に設定したままにしてください。

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

このオプションを使用して、特定のテーブルのサブセットを取り込むか、テーブルごとに宛先の命名をカスタマイズします。events テーブルは、以下に示すオプションの connector_options を受け入れます。その他のテーブルでは、テーブルごとのオプションは宣言されません。

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

YAML
resources:
  pipelines:
    amplitude_pipeline:
      name: amplitude_pipeline
      catalog: 'main'
      target: 'amplitude_data'
      channel: PREVIEW
      ingestion_definition:
        connection_name: amplitude_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'events'
              destination_catalog: 'main'
              destination_schema: 'amplitude_data'
              destination_table: 'events'
              connector_options:
                api_source_connector_options:
                  options:
                    start_datetime: '<start-datetime>'
                    request_time_range: '<request-time-range>'
          - table:
              source_schema: 'default'
              source_table: 'events_list'
              destination_catalog: 'main'
              destination_schema: 'amplitude_data'
              destination_table: 'events_list'
          - table:
              source_schema: 'default'
              source_table: 'cohorts'
              destination_catalog: 'main'
              destination_schema: 'amplitude_data'
              destination_table: 'cohorts'
          - table:
              source_schema: 'default'
              source_table: 'annotations'
              destination_catalog: 'main'
              destination_schema: 'amplitude_data'
              destination_table: 'annotations'

スキーマ全体を取り込む​

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

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

YAML
resources:
  pipelines:
    amplitude_pipeline:
      name: amplitude_pipeline
      catalog: 'main'
      target: 'amplitude_data'
      channel: PREVIEW
      ingestion_definition:
        connection_name: amplitude_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'amplitude_data'

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

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

YAML
resources:
  jobs:
    amplitude_job:
      name: amplitude_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: amplitude_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.amplitude_pipeline.id}

一般的なパターン​

高度なパイプライン構成については、「マネージド取り込みパイプラインの一般的なパターン」を参照してください。

次のステップ​

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

その他のリソース​

このページの見出し