メタデータのインポート用にカスタム コネクタを開発する

このドキュメントには、サードパーティ ソースからメタデータを抽出するカスタム コネクタを構築するためのリファレンス テンプレートが掲載されています。コネクタは、メタデータを Dataplex Universal Catalog にインポートするマネージド接続パイプラインを実行するときに使用します。

サードパーティ ソースからメタデータを抽出するコネクタを構築できます。たとえば、MySQL、SQL Server、Oracle、Snowflake、Databricks などのソースからデータを抽出するコネクタを構築できます。

このドキュメントのコネクタの例を参考に、独自のコネクタを構築します。このコネクタの例は、Oracle Database Express Edition（XE）データベースに接続します。このコネクタは Python で構築されていますが、Java、Scala、R も使用できます。

コネクタの仕組み

コネクタは、サードパーティ データソースからメタデータを抽出して、メタデータを Dataplex Universal Catalog ImportItem 形式に変換し、Dataplex Universal Catalog によってインポートできるメタデータ インポート ファイルを生成します。

このコネクタは、マネージド接続パイプラインの一部です。マネージド接続パイプラインは、Dataplex Universal Catalog のメタデータをインポートするために使用するオーケストレートされたワークフローです。マネージド接続パイプラインは、コネクタを実行し、インポート ワークフローの他のタスク（メタデータ インポート ジョブの実行、ログのキャプチャなど）を実行します。

マネージド接続パイプラインは、Google Cloud Serverless for Apache Spark バッチジョブを使用してコネクタを実行します。Serverless for Apache Spark は、サーバーレスの Spark 実行環境を提供します。Spark を使用しないコネクタを構築することもできますが、Spark を使用することをおすすめします。これにより、コネクタのパフォーマンスが向上します。

コネクタの要件

コネクタには次の要件があります。

  • コネクタは、Serverless for Apache Spark で実行できる Artifact Registry イメージであることが必要です。
  • コネクタは、Dataplex Universal Catalog メタデータ インポート ジョブ（metadataJobs.create API メソッド）によってインポートできる形式のメタデータ ファイルを生成する必要があります。詳細な要件については、メタデータのインポート ファイルをご覧ください。

  • パイプラインから情報を受信するには、コネクタが次のコマンドライン引数を受け入れるようにする必要があります。

    コマンドライン引数 パイプラインが指定する値
    target_project_id PROJECT_ID
    target_location_id REGION
    target_entry_group_id ENTRY_GROUP_ID
    output_bucket CLOUD_STORAGE_BUCKET_ID
    output_folder FOLDER_ID

    コネクタは、これらの引数を使用して、ターゲット エントリ グループ projects/PROJECT_ID/locations/REGION/entryGroups/ENTRY_GROUP_ID でメタデータを生成し、Cloud Storage バケット gs://CLOUD_STORAGE_BUCKET_ID/FOLDER_ID に書き込みます。 パイプラインを実行するたびに、バケット CLOUD_STORAGE_BUCKET_ID に新しいフォルダ FOLDER_ID が作成されます。コネクタは、メタデータのインポート ファイルをこのフォルダに書き込む必要があります。

パイプライン テンプレートは PySpark コネクタをサポートしています。テンプレートでは、ドライバ（mainPythonFileUri）が main.py という名前のコネクタ イメージ上のローカル ファイルであると想定しています。Spark コネクタ、別のドライバ URI、その他のオプションなど、他のシナリオに合わせてパイプライン テンプレートを変更できます。

PySpark を使用してメタデータのインポート ファイルにインポート アイテムを作成する方法は次のとおりです。

"""PySpark schemas for the data."""
entry_source_schema = StructType([
      StructField("display_name", StringType()),
      StructField("source", StringType())])

aspect_schema = MapType(StringType(),
                        StructType([
                            StructField("aspect_type", StringType()),
                            StructField("data", StructType([
                            ]))
                          ])
                        )

entry_schema = StructType([
  StructField("name", StringType()),
  StructField("entry_type", StringType()),
  StructField("fully_qualified_name", StringType()),
  StructField("parent_entry", StringType()),
  StructField("entry_source", entry_source_schema),
  StructField("aspects", aspect_schema)
])

import_item_schema = StructType([
  StructField("entry", entry_schema),
  StructField("aspect_keys", ArrayType(StringType())),
  StructField("update_mask", ArrayType(StringType()))
])

始める前に

このガイドは、Python と PySpark の知識があることを前提としています。

次の情報を確認します。

次の操作を行います。すべてのリソースを同じ Google Cloudロケーションに作成します。

  1. Create or select a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

  2. Verify that billing is enabled for your Google Cloud project.

  3. Enable the Dataplex, Dataproc, Workflows, and Artifact Registry APIs:

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles. 

    gcloud services enable dataplex.googleapis.com dataproc.googleapis.com workflows.googleapis.com artifactregistry.googleapis.com

  4. Install the Google Cloud CLI.

  5. 外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

  6. gcloud CLI を初期化するには、次のコマンドを実行します。

    gcloud init

  7. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/resourcemanager.projectCreator, roles/billing.projectManager, roles/serviceusage.admin, roles/iam.serviceAccountCreator, roles/iam.securityAdmin, roles/storage.admin, roles/artifactregistry.writer, roles/dataplex.entryGroupOwner, roles/dataplex.entryOwner, roles/dataplex.aspectTypeOwner 

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER"