カスタム パイプラインを使用してメタデータをインポートする

このドキュメントでは、メタデータ インポート API メソッドと独自のパイプラインを使用して、サードパーティ システムから Dataplex に Dataplex Catalog メタデータをインポートする方法について説明します。Dataplex Catalog メタデータは、エントリとそのアスペクトで構成されます。

Google Cloud 管理のオーケストレーション パイプラインを使用してメタデータを抽出してインポートする場合は、マネージド接続パイプラインを使用することをおすすめします。マネージド接続パイプラインでは、メタデータを抽出し、メタデータ インポート API メソッド(メタデータ インポート ファイル)の入力として使用できる形式で出力を生成する独自のコネクタを用意します。次に、Workflows を使用してパイプライン タスクをオーケストレートします。

手順の概要

metadata import API を使用してメタデータをインポートする、おおまかな手順は次のとおりです。

  1. ジョブのスコープを決定します。

    また、Dataplex がエントリとアスペクトに比較ロジックと同期モードを適用する方法についても説明します。

  2. インポートするデータを定義するメタデータのインポート ファイルを 1 つ以上作成します。

  3. メタデータのインポート ファイルを Cloud Storage バケットに保存します。

  4. メタデータのインポート ジョブを実行します。

このページの手順は、エントリ グループ、エントリタイプ、アスペクト タイプなど、Dataplex Catalog のコンセプトに精通していることを前提としています。詳細については、Dataplex Catalog の概要をご覧ください。

始める前に

メタデータをインポートする前に、このセクションのタスクを完了します。

必要なロール

Dataplex サービス アカウントに Cloud Storage バケットへのアクセスに必要な権限が付与されるようにするには、Dataplex サービス アカウントにバケットに対するストレージ オブジェクト閲覧者(roles/storage.objectViewer)IAM ロールと storage.buckets.get 権限を付与するよう管理者に依頼してください。

メタデータ ジョブを管理するために必要な権限を取得するには、次の IAM ロールを付与するよう管理者に依頼してください。

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

必要な権限は、カスタムロールや他の事前定義ロールから取得することもできます。

Google Cloud リソースを作成する

次の Google Cloud リソースを準備します。

  1. インポートするエントリのエントリ グループを作成します。
  2. インポートするアスペクトのアスペクト タイプを作成します。
  3. インポートするエントリのエントリタイプを作成します。
  4. メタデータのインポート ファイルを保存する Cloud Storage バケットを作成します。

メタデータ ジョブのコンポーネント

メタデータをインポートする際は、メタデータ ジョブの次のコンポーネントを考慮してください。

  • ジョブのスコープ: ジョブに設定するエントリ グループ、エントリタイプ、アスペクト タイプ。
  • 同期モード: ジョブ内のエントリとアスペクトに対して完全な更新と増分更新のどちらを行うか。
  • メタデータ インポート ファイル: ジョブ内のエントリとアスペクトに設定する値を定義するファイル。同じメタデータ ジョブで複数のメタデータ インポート ファイルを指定できます。ファイルを Cloud Storage に保存します。
  • 比較ロジック: Dataplex が変更するエントリとアスペクトを決定する方法。

ジョブのスコープ

ジョブ スコープでは、エントリ グループ、エントリ タイプ、必要に応じてメタデータ ジョブに配置するアスペクト タイプを定義します。メタデータをインポートした際に、ジョブのスコープ内のリソースに属するエントリとアスペクトを変更します。

ジョブのスコープを定義するには、次のガイドラインに従ってください。

  • エントリ グループ: ジョブに配置する単一のエントリ グループを指定します。このジョブは、このエントリ グループに属するエントリのみを変更します。エントリ グループとジョブは同じリージョンに存在する必要があります。

  • エントリタイプ: ジョブに設定する 1 つ以上のエントリタイプを指定します。ジョブは、これらのエントリタイプに属するエントリのみを変更します。エントリタイプのロケーションは、ジョブのロケーションと一致しているか、エントリタイプがグローバルであることが必要です。

  • アスペクト タイプ: 省略可。ジョブに設定するアスペクト タイプを 1 つ以上指定します。アスペクト タイプのスコープを指定すると、ジョブはそれらのアスペクト タイプに属するアスペクトのみを変更します。アスペクト タイプのロケーションは、ジョブのロケーションと一致しているか、アスペクト タイプがグローバルであることが必要です。

ジョブスコープは、メタデータ ジョブを作成するときに指定します。

同期モード

同期モードでは、メタデータ ジョブ内のエントリとアスペクトに対して完全な更新と増分更新のどちらを実行するかを指定します。

  • FULL: エントリでサポートされています。ジョブのスコープのすべてのエントリが変更されます。

    Dataplex にエントリが存在しても、メタデータ インポート ファイルに含まれていない場合は、メタデータ ジョブの実行時にエントリが削除されます。

    アスペクトの完全同期はサポートされていません。

  • INCREMENTAL: アスペクトでサポートされています。アスペクトが変更されるのは、メタデータ インポート ファイルの updateMask フィールドと aspectKeys フィールドにアスペクトへの参照が含まれている場合のみです。メタデータのインポート ファイルのこれらのフィールドの詳細については、このドキュメントのインポート アイテムの構造をご覧ください。

    エントリの増分同期はサポートされていません。

同期モードは、メタデータ ジョブを作成するときに指定します。

メタデータのインポート ファイル

メタデータのインポート ファイルは、変更するエントリとアスペクトのコレクションです。これらのエントリとアスペクトに属するすべてのフィールドに設定する値を定義します。メタデータ ジョブを実行する前にファイルを準備します。

次の一般的なガイドラインが適用されます。

  • 同じメタデータ ジョブで複数のメタデータ インポート ファイルを指定できます。
  • ファイルで指定したエントリは、ジョブのスコープに含まれるリソースの既存のエントリをすべて完全に置き換えます。つまり、追加または更新する値だけでなく、ジョブ内のすべてのエントリの値を配置する必要があります。開始点として使用するプロジェクト内の現在のエントリのリストを取得するには、entries.list API メソッドを使用します。

  • メタデータ ジョブの一部としてメタデータのインポート ファイルを指定する必要があります。ジョブのスコープに含まれるエントリの既存のデータをすべて削除する場合は、空のメタデータ インポート ファイルを指定します。

  • ファイルに配置するエントリとアスペクトはすべて、ジョブのスコープで定義したエントリグループ、エントリタイプ、アスペクトタイプに属している必要があります。

次のセクションの詳細なガイドラインに沿って、メタデータ インポート ファイルを作成します。

ファイルの構造

メタデータ インポート ファイルの各行には、1 つのインポート アイテムに対応する JSON オブジェクトが含まれています。インポート アイテムは、エントリとそれに関連付けられたアスペクトの変更値を記述するオブジェクトです。

1 つのメタデータ インポート ファイルに複数のインポート アイテムを指定できます。ただし、メタデータ ジョブで同じインポート アイテムを複数回指定しないでください。各インポート アイテムは改行文字(0x0a)で区切ります。

各インポート アイテムを改行文字で区切ったメタデータ インポート ファイルは、次の例のようになります。

{ "entry": { "name": "entry 1", #Information about entry 1 }
{ "entry": { "name": "entry 2", #Information about entry 2 }

インポート アイテムの構造

メタデータ インポート ファイル内の各インポート アイテムには、次のフィールドが含まれます(ImportItem を参照)。次の例では、読みやすくするために改行文字が使用されていますが、ファイルを保存するときは、各インポート アイテムの後にのみ改行文字を含めます。1 つのインポート アイテムのフィールドの間には改行を含めないでください。

{
  "entry": {
    "name": "ENTRY_NAME",
    "entryType": "ENTRY_TYPE",
    "entrySource": {
      "resource": "RESOURCE",
      "system": "SYSTEM",
      "platform": "PLATFORM",
      "displayName": "DISPLAY_NAME",
      "description": "DESCRIPTION",
      "createTime": "ENTRY_CREATE_TIMESTAMP",
      "updateTime": "ENTRY_UPDATE_TIMESTAMP"
    }
    "aspects": {
      "ASPECT": {
        "data": {
          "KEY": "VALUE"
        },
        "aspectSource": {
          "createTime": "ASPECT_CREATE_TIMESTAMP"
          "updateTime": "ASPECT_UPDATE_TIMESTAMP"
        }
      },
      # Additional aspect maps
    },
    "parentEntry": "PARENT_ENTRY",
    "fullyQualifiedName": "FULLY_QUALIFIED_NAME"
  },
  "updateMask": "UPDATE_MASK_FIELDS",
  "aspectKeys": [
    "ASPECT_KEY",
    # Additional aspect keys
  ],
}

次のように置き換えます。

  • entry: エントリとそれに関連付けられたアスペクトに関する情報:

    • ENTRY_NAME: エントリの相対的なリソース名(projects/PROJECT_ID_OR_NUMBER/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID/entries/ENTRY_ID の形式)。
    • ENTRY_TYPE: このエントリの作成に使用されたエントリタイプの相対的なリソース名(projects/PROJECT_ID_OR_NUMBER/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID 形式)。
    • entrySource: エントリによって表されるデータリソースに関するソースシステムの情報。
      • RESOURCE: ソースシステムのリソースの名前。
      • SYSTEM: ソースシステムの名前。
      • PLATFORM: ソースシステムを含むプラットフォーム。
      • DISPLAY_NAME: ユーザー フレンドリーな表示名。
      • DESCRIPTION: エントリの説明。
      • ENTRY_CREATE_TIMESTAMP: エントリがソースシステムで作成された時刻。
      • ENTRY_UPDATE_TIMESTAMP: ソースシステムでエントリが更新された時刻。
    • aspects: エントリにアタッチされているアスペクト。aspect オブジェクトとそのデータは、アスペクト マップと呼ばれます。

      • ASPECT: エントリにアタッチされているアスペクト。アスペクトがエントリにどのようにアタッチされているかに応じて、次のいずれかの形式を使用します。

        • アスペクトがエントリに直接アタッチされている場合は、そのアスペクト タイプの相対的なリソース名を PROJECT_ID_OR_NUMBER.LOCATION_ID.ASPECT_TYPE_ID の形式で指定します。
        • アスペクトがエントリのパスにアタッチされている場合は、アスペクト タイプのパスを PROJECT_ID_OR_NUMBER.LOCATION_ID.ASPECT_TYPE_ID@PATH の形式で指定します。
      • KEYVALUE: アスペクト タイプ メタデータ テンプレートに基づくアスペクトのコンテンツ。コンテンツは UTF-8 としてエンコードする必要があります。フィールドの最大サイズは 120 KB です。data ディクショナリは、空であっても必須です。

      • ASPECT_CREATE_TIMESTAMP: 移行元システムでアスペクトが作成された時刻。

      • ASPECT_UPDATE_TIMESTAMP: ソースシステムでアスペクトが更新された時刻。

    • PARENT_ENTRY: 親エントリのリソース名。

    • FULLY_QUALIFIED_NAME: 外部システムから参照できるエントリの名前。完全修飾名をご覧ください。

  • UPDATE_MASK_FIELDS: 更新するフィールド(Entry リソースを基準としたパス)。各フィールドはカンマで区切ります。

    FULL エントリの同期モードでは、Dataplex には、アスペクトを含む、変更可能なエントリのすべてのフィールドのパスが含まれます。

    エントリの作成または再作成時に、updateMask フィールドは無視されます。

  • ASPECT_KEY: 変更するアスペクト。次の構文をサポートします。

    • ASPECT_TYPE_REFERENCE: エントリに直接アタッチされているアスペクトのアスペクト タイプと一致します。
    • ASPECT_TYPE_REFERENCE@PATH: アスペクト タイプと指定されたパスと一致します。
    • ASPECT_TYPE_REFERENCE@*: すべてのパスのアスペクト タイプと一致します。

    ASPECT_TYPE_REFERENCE は、アスペクト タイプへの参照に置き換えます。形式は PROJECT_ID_OR_NUMBER.LOCATION_ID.ASPECT_TYPE_ID です。

    このフィールドを空のままにすると、指定されたエントリ内に存在するアスペクトを正確に指定したものとして扱われます。

    FULL エントリの同期モードでは、Dataplex がエントリに必要なすべてのアスペクトのキーを暗黙的に追加します。

ファイルの要件

メタデータ インポート ファイルには次の要件があります。

  • ファイルは、改行区切りの JSON ファイルである JSON Lines ファイル形式でなければなりません。各インポート アイテムは改行文字(0x0a)で区切ります。
  • ファイルは UTF-8 文字エンコードを使用する必要があります。
  • サポートされているファイル拡張子は .jsonl.json です。
  • 各メタデータ インポート ファイルのファイルサイズは 1 GiB 未満にする必要があります。メタデータ ジョブ内のすべてのデータの最大合計サイズは 3 GB です。これには、ジョブに関連付けられたすべてのファイルとメタデータが含まれます。
  • ファイルで指定するエントリとアスペクトは、メタデータ ジョブのスコープに含まれる必要があります。
  • ファイルは Cloud Storage バケットにアップロードする必要があります。ファイルを CLOUD_STORAGE_URI/deletions/ という名前のフォルダに保存しないでください。

比較ロジック

Dataplex は、メタデータ インポート ファイルで指定された値とタイムスタンプを、プロジェクトに存在する値とタイムスタンプと比較して、変更するエントリとアスペクトを決定します。

概要として、メタデータ インポート ファイルで提案された変更が 1 つ以上、ジョブの実行時にプロジェクトの状態を変更する場合、Dataplex は古いデータを導入することなくプロジェクトの値を更新します。提案された変更は、メタデータ インポート ファイルの更新マスク フィールドまたはアスペクトキー フィールドで参照する必要があります。

ジョブのスコープに含まれるエントリごとに、Dataplex は次のいずれかの処理を行います。

  • エントリとアタッチされたアスペクトを作成します。メタデータ インポート ファイルに、プロジェクトに存在しないエントリが含まれている場合、Dataplex はエントリとアタッチされたアスペクトを作成します。
  • エントリとアタッチされたアスペクトを削除します。エントリがプロジェクトに存在するものの、メタデータ インポート ファイルにエントリが含まれていない場合、Dataplex はエントリとエントリにアタッチされたアスペクトをプロジェクトから削除します。
  • エントリとアタッチされたアスペクトを更新します。エントリがメタデータ インポート ファイルとプロジェクトの両方に存在する場合、Dataplex はエントリソースのタイムスタンプとエントリに関連付けられているアスペクトソースのタイムスタンプを評価して、変更する値を決定します。次に、Dataplex は次のいずれかを行います。

    • エントリを再作成します。メタデータ インポート ファイルのエントリソースの作成タイムスタンプが、プロジェクト内の対応するタイムスタンプよりも新しい場合、Dataplex はプロジェクト内のエントリを再作成します。
    • エントリを更新します。メタデータ インポート ファイルのエントリソースの更新タイムスタンプが、プロジェクト内の対応するタイムスタンプよりも新しい場合、Dataplex はプロジェクト内のエントリを更新します。
    • アスペクトを作成します。アスペクトがプロジェクトに存在せず、メタデータ インポート ファイルのエントリ オブジェクト、更新マスク フィールド、アスペクトキー フィールドに含まれている場合、Dataplex はアスペクトを作成します。
    • アスペクトを削除します。アスペクトがプロジェクトに存在し、メタデータ インポート ファイルの更新マスク フィールドとアスペクトキー フィールドに含まれていても、エントリ オブジェクトに含まれていない場合、Dataplex はアスペクトを削除します。
    • アスペクトを更新します。プロジェクトにアスペクトが存在し、メタデータ インポート ファイルのエントリ オブジェクト、更新マスク フィールド、アスペクトキー フィールドに含まれていて、メタデータ インポート ファイルのアスペクト ソースの更新タイムスタンプがプロジェクト内の対応するタイムスタンプよりも新しい場合、Dataplex はアスペクトを更新します。

      メタデータ インポート ファイルにアスペクト ソースの更新タイムスタンプが指定されていないが、対応するエントリが更新対象としてマークされている場合、Dataplex はアスペクトも更新します。

      ただし、メタデータ インポート ファイルの少なくとも 1 つのアスペクトにプロジェクト内の対応するタイムスタンプよりも古いタイムスタンプがある場合、Dataplex はアタッチされたエントリを更新しません。

メタデータのインポート ファイルを作成する

メタデータをインポートする前に、ジョブのメタデータ インポート ファイルを作成します。手順は次のとおりです。

  1. このドキュメントで前述したガイドラインに沿って、メタデータ インポート ファイルを準備します。
  2. ファイルを Cloud Storage バケットにアップロードします

同じメタデータ ジョブで複数のメタデータ インポート ファイルを指定できます。複数のファイルを指定するには、同じ Cloud Storage バケットにファイルを保存します。ジョブを実行するときに、特定のファイルではなくバケットを指定します。Dataplex は、サブフォルダ内のファイルを含む、バケットに保存されているすべてのファイルからメタデータをインポートします。

メタデータのインポート ジョブを実行する

メタデータのインポート ファイルを作成したら、API を使用してメタデータのインポート ジョブを実行します。

REST

メタデータをインポートするには、metadataJobs.create メソッドを使用します。

リクエストのデータを使用する前に、次のように置き換えます。

  • PROJECT_NUMBER: Google Cloud プロジェクト番号またはプロジェクト ID。
  • LOCATION_ID: Google Cloud のロケーション(us-central1 など)。
  • METADATA_JOB_ID: 省略可。メタデータ ジョブ ID。
  • CLOUD_STORAGE_URI: メタデータのインポート ファイルを含む Cloud Storage バケットまたはフォルダの URI。ファイルの要件の詳細については、メタデータのインポート ファイルをご覧ください。

  • ENTRY_GROUP: ジョブのスコープ内にあるエントリ グループの相対的なリソース名。形式は projects/PROJECT_ID_OR_NUMBER/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID です。エントリ グループを 1 つのみ指定します。詳細については、ジョブのスコープをご覧ください。
  • ENTRY_TYPE: ジョブのスコープ内にあるエントリ タイプの相対的なリソース名。形式は projects/PROJECT_ID_OR_NUMBER/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID です。詳細については、ジョブのスコープをご覧ください。

  • ASPECT_TYPE: 省略可。ジョブのスコープ内にあるアスペクト タイプの相対的なリソース名(projects/PROJECT_ID_OR_NUMBER/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID 形式)。詳細については、ジョブのスコープをご覧ください。
  • LOG_LEVEL: キャプチャするログのレベル(INFODEBUG など)。詳細については、ジョブのログを表示してトラブルシューティングを行うをご覧ください。

HTTP メソッドと URL:

POST https://dataplex.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION_ID/metadataJobs?metadataJobId=METADATA_JOB_ID

リクエストの本文(JSON):

{
  "type": IMPORT,
  "import_spec": {
    "source_storage_uri": gs://CLOUD_STORAGE_URI/,
    "scope": {
      "entryGroups": [
        ENTRY_GROUP
      ],
      "entry_types": [
        ENTRY_TYPE
      ],
      "aspect_types": [
        ASPECT_TYPE
      ]
    },
    "entry_sync_mode": FULL,
    "aspect_sync_mode": INCREMENTAL,
    "log_level": LOG_LEVEL
  }
}

リクエストを送信するには、次のいずれかのオプションを展開します。

レスポンスで長時間実行オペレーションを識別できます。

メタデータ ジョブの詳細情報を取得する

メタデータ ジョブに関する情報(ジョブのステータス、変更されたエントリの数など)を取得する手順は次のとおりです。失敗したジョブのトラブルシューティング方法の詳細については、このドキュメントのジョブのログを表示してトラブルシューティングを行うセクションをご覧ください。

REST

メタデータ ジョブに関する情報を取得するには、metadataJobs.get メソッドを使用します。

メタデータ ジョブのリストを取得する

最新のメタデータ ジョブのリストを取得できます。終了状態に達した古いジョブは、定期的にシステムから削除されます。

REST

最新のメタデータ ジョブのリストを取得するには、metadataJobs.list メソッドを使用します。

メタデータ ジョブをキャンセルする

実行する必要がないメタデータ ジョブはキャンセルできます。

REST

メタデータ ジョブをキャンセルするには、metadataJobs.cancel メソッドを使用します。

ジョブのログを表示してトラブルシューティングを行う

Cloud Logging を使用して、メタデータ ジョブのログを表示します。詳細については、Dataplex ログをモニタリングするをご覧ください。

ログレベルは、メタデータ ジョブを作成するときに構成します。次のログレベルを使用できます。

  • INFO: ジョブ全体のレベルでログを提供します。インポート アイテムに関する集計ログが含まれますが、エラーのあるインポート アイテムは指定されません。
  • DEBUG: インポート アイテムごとに詳細なログを提供します。デバッグレベルのロギングを使用して、特定のインポート アイテムに関する問題のトラブルシューティングを行います。たとえば、デバッグレベルのロギングを使用して、ジョブスコープにないリソース、関連するエントリタイプまたはアスペクトタイプに準拠していないエントリまたはアスペクト、メタデータ インポート ファイルのその他の構成ミスを特定します。

検証エラー

Dataplex は、プロジェクトの現在のメタデータと照らし合わせてメタデータ インポート ファイルを検証します。検証の問題がある場合、ジョブのステータスは次のいずれかの状態を返す可能性があります。

  • FAILED: メタデータ インポート ファイルにエラーがある場合に発生する。Dataplex はメタデータをインポートせず、ジョブは失敗します。メタデータ インポート ファイルのエラーの例を以下に示します。
    • ファイル内の項目を有効なインポート項目に解析できない。
    • ファイル内のエントリまたはアスペクトが、ジョブのスコープに含まれないエントリグループ、エントリタイプ、またはアスペクトタイプに属している。
    • ジョブで同じエントリ名が複数回指定されている。
    • アスペクト マップまたはアスペクト キーで指定されたアスペクト タイプが、PROJECT_ID_OR_NUMBER.LOCATION_ID.ASPECT_TYPE_ID@OPTIONAL_PATH の形式を使用していない。
  • SUCCEEDED_WITH_ERRORS: メタデータ インポート ファイルは正常に解析できるものの、ファイル内のアイテムをインポートすると、プロジェクト内のエントリが不整合状態になる場合に発生する。Dataplex はこのようなエントリを無視しますが、残りのメタデータをファイルからインポートします。

ジョブのログを使用してエラーのトラブルシューティングを行います。

次のステップ