开发用于元数据导入的自定义连接器

本文档提供了一个参考模板，可供您构建自定义连接器，以从第三方来源提取元数据。在运行将元数据导入 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. 如果您使用的是外部身份提供方 (IdP)，则必须先使用联合身份登录 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" --role=ROLE

    Replace the following:

    • PROJECT_ID: Your project ID.
    • USER_IDENTIFIER: The identifier for your user account. For example, myemail@example.com.
    • ROLE: The IAM role that you grant to your user account.

  8. Set up authentication:

    1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator) and the Project IAM Admin role (roles/resourcemanager.projectIamAdmin). Learn how to grant roles.

    2. Create the service account: 

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      Replace SERVICE_ACCOUNT_NAME with a name for the service account.

    3. Grant the roles/owner IAM role to the service account: 

      gcloud projects