此页面由 Cloud Translation API 翻译。

使用工作流从自定义来源导入元数据
使用集合让一切井井有条根据您的偏好保存内容并对其进行分类。

本文档介绍了如何通过在 Workflows 中运行受管连接流水线，将来自第三方来源的元数据导入 Dataplex。

如需设置托管式连接流水线，您需要为数据源构建连接器。然后在 Workflows 中运行该流水线。该流水线会从您的数据源提取元数据，然后将元数据导入 Dataplex。如有必要，流水线还会在您的 Google Cloud 项目中创建 BigQuery 通用目录条目组。

如需详细了解托管式连接，请参阅托管式连接概览。

准备工作

在导入元数据之前，请完成本部分中的任务。

构建连接器

连接器会从您的数据源提取元数据，并生成可由 Dataplex 导入的元数据导入文件。该连接器是一个可在 Dataproc Serverless 上运行的 Artifact Registry 映像。

构建自定义连接器，用于从第三方来源提取元数据。

如需查看可用作参考模板来构建您自己的连接器的示例连接器，请参阅开发用于元数据导入的自定义连接器。

配置 Google Cloud 资源

Enable the Workflows, Dataproc, Cloud Storage, Dataplex, Secret Manager, Artifact Registry, and Cloud Scheduler APIs.
Enable the APIs

如果您不打算按时间表运行流水线，则无需启用 Cloud Scheduler API。
在 Secret Manager 中创建 Secret，以存储第三方数据源的凭证。
配置虚拟私有云 (VPC) 网络以运行 Dataproc Serverless for Spark 工作负载。
创建 Cloud Storage 存储桶以存储元数据导入文件。
创建以下通用目录资源：
1. 为要导入的条目创建自定义切面类型。
2. 为要导入的条目创建自定义条目类型。

所需的角色

服务账号代表工作流的身份，并决定了工作流拥有的权限以及它可以访问的 Google Cloud 资源。您需要一个 Workflows 服务账号（用于运行流水线）和一个 Dataproc Serverless 服务账号（用于运行连接器）。

您可以使用 Compute Engine 默认服务账号 (PROJECT_NUMBER-compute@developer.gserviceaccount.com)，也可以创建自己的服务账号来运行托管式连接流水线。

控制台

在 Google Cloud 控制台中，前往 IAM 页面。

转到 IAM
选择要导入元数据的目标项目。
点击 授予访问权限，然后输入服务账号的邮箱。
为服务账号分配以下角色：
- Logs Writer
- Dataplex Entry Group Owner
- Dataplex Metadata Job Owner
- Dataplex Catalog Editor
- Dataproc Editor
- Dataproc Worker
- Secret Manager Secret Accessor -针对存储数据源凭证的 Secret
- Storage Object User - 针对 Cloud Storage 存储桶
- Artifact Registry Reader - 针对包含连接器映像的 Artifact Registry 存储库
- Service Account User - 如果使用不同的服务账号，请为运行 Workflows 的账号授予此角色，以便其可调用运行 Dataproc Serverless 批处理作业的服务账号
- Workflows Invoker - 如果您想安排流水线
保存更改。

gcloud

向服务账号授予角色。运行以下命令：

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/logging.logWriter
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.entryGroupOwner
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.metadataJobOwner
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataplex.catalogEditor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataproc.editor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/dataproc.worker

替换以下内容：

PROJECT_ID：用于导入元数据的目标 Google Cloud项目名称。
SERVICE_ACCOUNT_ID：服务账号，例如 my-service-account@my-project.iam.gserviceaccount.com。

在资源级别为该服务账号授予以下角色：

gcloud secrets add-iam-policy-binding SECRET_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/secretmanager.secretaccessor
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role=roles/storage.objectUser \
    --condition=resource.name.startsWith('projects/_/buckets/BUCKET_ID')
gcloud artifacts repositories add-iam-policy-binding REPOSITORY \
    --location=REPOSITORY_LOCATION \
    --member=SERVICE_ACCOUNT_ID} \
    --role=roles/artifactregistry.reader

替换以下内容：

SECRET_ID：用于存储数据源凭证的 Secret 的 ID。其格式为 projects/PROJECT_ID/secrets/SECRET_ID。
BUCKET_ID：Cloud Storage 存储桶的名称。
REPOSITORY：包含连接器映像的 Artifact Registry 制品库。
REPOSITORY_LOCATION：存储库托管所在的 Google Cloud位置。

为运行 Workflows 的服务账号授予运行 Dataproc Serverless 批处理作业的服务账号的 roles/iam.serviceAccountUser 角色。即使您为 Workflows 和 Dataproc Serverless 使用相同的服务账号，也必须授予此角色。
```
gcloud iam service-accounts add-iam-policy-binding \
    serviceAccount:SERVICE_ACCOUNT_ID \
    --member='SERVICE_ACCOUNT_ID' \
    --role='roles/iam.serviceAccountUser'
```
如果您使用的是不同的服务账号，--member 标志的值就是运行 Dataproc Serverless 批量作业的服务账号。

如果您想安排流水线，请为服务账号授予以下角色：

gcloud projects add-iam-policy-binding PROJECT_ID \
 --member="SERVICE_ACCOUNT_ID" \
 --role=roles/workflows.invoker

导入元数据

如需导入元数据，请创建并执行一个用于运行托管式连接性流水线的工作流。（可选）您还可以为流水线创建一个运行时间表。

控制台

创建工作流。提供以下信息：

服务账号：您在本文档的所需角色部分中配置的服务账号。
加密：选择 Google-managed encryption key。

注意：客户管理的加密密钥 (CMEK) 不会传播到 Dataplex 和 Dataproc Serverless 作业，因此无法实现端到端加密。

定义工作流：提供以下定义文件：

main:
  params: [args]
  steps:
    - init:
        assign:
        - WORKFLOW_ID: ${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
        - NETWORK_URI: ${default(map.get(args, "NETWORK_URI"), "")}
        - SUBNETWORK_URI: ${default(map.get(args, "SUBNETWORK_URI"), "")}
        - NETWORK_TAGS: ${default(map.get(args, "NETWORK_TAGS"), [])}

    - check_networking:
        switch:
          - condition: ${NETWORK_URI != "" and SUBNETWORK_URI != ""}
            raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
          - condition: ${NETWORK_URI == "" and SUBNETWORK_URI == ""}
            steps:
             - submit_extract_job_with_default_network_uri:
                  assign:
                    - NETWORK_TYPE: "networkUri"
                    - NETWORKING: ${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}  
          - condition: ${NETWORK_URI != ""}
            steps:
              - submit_extract_job_with_network_uri:
                  assign:
                    - NETWORKING: ${NETWORK_URI}
                    - NETWORK_TYPE: "networkUri"
          - condition: ${SUBNETWORK_URI != ""}
            steps:
              - submit_extract_job_with_subnetwork_uri:
                  assign:
                    - NETWORKING: ${SUBNETWORK_URI}
                    - NETWORK_TYPE: "subnetworkUri"
        next: check_create_target_entry_group

    - check_create_target_entry_group:
        switch:
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == true}
            next: create_target_entry_group
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == false}
            next: generate_extract_job_link

    - create_target_entry_group:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: generate_extract_job_link

    - generate_extract_job_link:
        call: sys.log
        args:
            data: ${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
            severity: "INFO"
        next: submit_pyspark_extract_job

    - submit_pyspark_extract_job:
        call: http.post
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          headers:
            Content-Type: "application/json"
          query:
            batchId: ${WORKFLOW_ID}
          body:
            pysparkBatch:
              mainPythonFileUri: file:///main.py
              args:
                - ${"--target_project_id=" + args.TARGET_PROJECT_ID}
                - ${"--target_location_id=" + args.CLOUD_REGION}
                - ${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                - ${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                - ${"--output_folder=" + WORKFLOW_ID}
                - ${args.ADDITIONAL_CONNECTOR_ARGS}
            runtimeConfig:
                containerImage: ${args.CUSTOM_CONTAINER_IMAGE}
            environmentConfig:
                executionConfig:
                    serviceAccount: ${args.SERVICE_ACCOUNT}
                    stagingBucket: ${args.CLOUD_STORAGE_BUCKET_ID}
                    ${NETWORK_TYPE}: ${NETWORKING}
                    networkTags: ${NETWORK_TAGS}
        result: RESPONSE_MESSAGE
        next: check_pyspark_extract_job

    - check_pyspark_extract_job:
        call: http.get
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: PYSPARK_EXTRACT_JOB_STATUS
        next: check_pyspark_extract_job_done

    - check_pyspark_extract_job_done:
        switch:
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
            next: generate_import_logs_link
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
            raise: ${PYSPARK_EXTRACT_JOB_STATUS}
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
            raise: ${PYSPARK_EXTRACT_JOB_STATUS}
        next: pyspark_extract_job_wait

    - pyspark_extract_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_pyspark_extract_job

    - generate_import_logs_link:
        call: sys.log
        args:
            data: ${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
            severity: "INFO"
        next: submit_import_job

    - submit_import_job:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          body:
            type: IMPORT
            import_spec:
              source_storage_uri: ${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
              entry_sync_mode: FULL
              aspect_sync_mode: INCREMENTAL
              log_level: ${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
              scope:
                entry_groups: 
                  - ${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                entry_types: ${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                aspect_types: ${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
        result: IMPORT_JOB_RESPONSE
        next: get_job_start_time

    - get_job_start_time:
        assign:
          - importJobStartTime: ${sys.now()}
        next: import_job_startup_wait

    - import_job_startup_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: initial_get_import_job

    - initial_get_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_status_available

    - check_import_job_status_available:
        switch:
          - condition: ${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
            next: kill_import_job
          - condition: ${"status" in IMPORT_JOB_STATUS.body}
            next: check_import_job_done
        next: import_job_status_wait

    - import_job_status_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_import_job_status_available

    - check_import_job_done:
        switch:
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
            next: the_end
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
            next: kill_import_job
        next: import_job_wait

    - get_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_done

    - import_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: get_import_job

    - kill_import_job:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: get_killed_import_job

    - get_killed_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: KILLED_IMPORT_JOB_STATUS
        next: killed

    - killed:
        raise: ${KILLED_IMPORT_JOB_STATUS}

    - the_end:
        return: ${IMPORT_JOB_STATUS}

如需按需运行流水线，请执行工作流。

提供以下运行时参数：
```
{
    "TARGET_PROJECT_ID": "PROJECT_ID",
    "CLOUD_REGION": "LOCATION_ID",
    "TARGET_ENTRY_GROUP_ID": "ENTRY_GROUP_ID",
    "CREATE_TARGET_ENTRY_GROUP": CREATE_ENTRY_GROUP_BOOLEAN,
    "CLOUD_STORAGE_BUCKET_ID": "BUCKET_ID",
    "SERVICE_ACCOUNT": "SERVICE_ACCOUNT_ID",
    "ADDITIONAL_CONNECTOR_ARGS": [ADDITIONAL_CONNECTOR_ARGUMENTS],
    "CUSTOM_CONTAINER_IMAGE": "CONTAINER_IMAGE",
    "IMPORT_JOB_SCOPE_ENTRY_TYPES": [ENTRY_TYPES],
    "IMPORT_JOB_SCOPE_ASPECT_TYPES": [ASPECT_TYPES],
    "IMPORT_JOB_LOG_LEVEL": "INFO",
    "NETWORK_TAGS": [],
    "NETWORK_URI": "",
    "SUBNETWORK_URI": ""
 }
```
替换以下内容：
- PROJECT_ID：用于导入元数据的目标 Google Cloud项目名称。
- LOCATION_ID：Dataproc Serverless 和元数据导入作业的运行位置，也是元数据的导入目标 Google Cloud 位置。
- ENTRY_GROUP_ID：要导入元数据的条目组 ID。条目组 ID 可以包含小写字母、数字和连字符。
  
  此条目组的完整资源名称为 projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID。
- CREATE_ENTRY_GROUP_BOOLEAN：如果希望流水线在项目中创建尚未存在的条目组，请将此值设为 true。
- BUCKET_ID：用于存储连接器生成的元数据导入文件的 Cloud Storage 存储桶的名称。每次执行工作流都会创建一个新文件夹。
- SERVICE_ACCOUNT_ID：您在本文档的所需角色部分中配置的服务账号。服务账号在 Dataproc Serverless 中运行连接器。
- ADDITIONAL_CONNECTOR_ARGUMENTS：传递给连接器的附加参数列表。如需查看示例，请参阅开发用于元数据导入的自定义连接器。请使用双引号将每个参数括起来，并用英文逗号分隔各个参数。
- CONTAINER_IMAGE：托管在 Artifact Registry 中的连接器自定义容器映像。
- ENTRY_TYPES：导入范围内的条目类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
- ASPECT_TYPES：导入范围内的方面类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
- 可选：对于 NETWORK_TAGS 参数，提供网络标记列表。
- 可选：对于 NETWORK_URI 参数，请提供连接到数据源的 VPC 网络的 URI。如果您提供网络，请省略子网参数。
- 可选：对于 SUBNETWORK_URI 参数，请提供连接到数据源的子网 URI。如果您提供了子网，请省略网络参数。
根据导入的元数据量不同，运行流水线可能需要几分钟或更长时间。如需详细了解如何查看进度，请参阅访问工作流执行结果。

流水线运行完毕后，您可以在通用目录中搜索导入的元数据。
可选：如果您想按计划运行流水线，请使用 Cloud Scheduler 创建时间表。提供以下信息：
- 频率：用于定义流水线运行时间表的 unix-cron 表达式。
- 工作流参数：连接器的运行时参数，如上一步中所述。
- 服务账号：服务账号。该服务账号用于管理调度器。

gcloud

将以下工作负载定义保存为 YAML 文件：

main:
  params: [args]
  steps:
    - init:
        assign:
        - WORKFLOW_ID: ${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
        - NETWORK_URI: ${default(map.get(args, "NETWORK_URI"), "")}
        - SUBNETWORK_URI: ${default(map.get(args, "SUBNETWORK_URI"), "")}
        - NETWORK_TAGS: ${default(map.get(args, "NETWORK_TAGS"), [])}

    - check_networking:
        switch:
          - condition: ${NETWORK_URI != "" and SUBNETWORK_URI != ""}
            raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
          - condition: ${NETWORK_URI == "" and SUBNETWORK_URI == ""}
            steps:
             - submit_extract_job_with_default_network_uri:
                  assign:
                    - NETWORK_TYPE: "networkUri"
                    - NETWORKING: ${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}  
          - condition: ${NETWORK_URI != ""}
            steps:
              - submit_extract_job_with_network_uri:
                  assign:
                    - NETWORKING: ${NETWORK_URI}
                    - NETWORK_TYPE: "networkUri"
          - condition: ${SUBNETWORK_URI != ""}
            steps:
              - submit_extract_job_with_subnetwork_uri:
                  assign:
                    - NETWORKING: ${SUBNETWORK_URI}
                    - NETWORK_TYPE: "subnetworkUri"
        next: check_create_target_entry_group

    - check_create_target_entry_group:
        switch:
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == true}
            next: create_target_entry_group
          - condition: ${args.CREATE_TARGET_ENTRY_GROUP == false}
            next: generate_extract_job_link

    - create_target_entry_group:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: generate_extract_job_link

    - generate_extract_job_link:
        call: sys.log
        args:
            data: ${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
            severity: "INFO"
        next: submit_pyspark_extract_job

    - submit_pyspark_extract_job:
        call: http.post
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          headers:
            Content-Type: "application/json"
          query:
            batchId: ${WORKFLOW_ID}
          body:
            pysparkBatch:
              mainPythonFileUri: file:///main.py
              args:
                - ${"--target_project_id=" + args.TARGET_PROJECT_ID}
                - ${"--target_location_id=" + args.CLOUD_REGION}
                - ${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                - ${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                - ${"--output_folder=" + WORKFLOW_ID}
                - ${args.ADDITIONAL_CONNECTOR_ARGS}
            runtimeConfig:
                containerImage: ${args.CUSTOM_CONTAINER_IMAGE}
            environmentConfig:
                executionConfig:
                    serviceAccount: ${args.SERVICE_ACCOUNT}
                    stagingBucket: ${args.CLOUD_STORAGE_BUCKET_ID}
                    ${NETWORK_TYPE}: ${NETWORKING}
                    networkTags: ${NETWORK_TAGS}
        result: RESPONSE_MESSAGE
        next: check_pyspark_extract_job

    - check_pyspark_extract_job:
        call: http.get
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: PYSPARK_EXTRACT_JOB_STATUS
        next: check_pyspark_extract_job_done

    - check_pyspark_extract_job_done:
        switch:
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
            next: generate_import_logs_link
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
            raise: ${PYSPARK_EXTRACT_JOB_STATUS}
          - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
            raise: ${PYSPARK_EXTRACT_JOB_STATUS}
        next: pyspark_extract_job_wait

    - pyspark_extract_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_pyspark_extract_job

    - generate_import_logs_link:
        call: sys.log
        args:
            data: ${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
            severity: "INFO"
        next: submit_import_job

    - submit_import_job:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          body:
            type: IMPORT
            import_spec:
              source_storage_uri: ${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
              entry_sync_mode: FULL
              aspect_sync_mode: INCREMENTAL
              log_level: ${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
              scope:
                entry_groups: 
                  - ${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                entry_types: ${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                aspect_types: ${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
        result: IMPORT_JOB_RESPONSE
        next: get_job_start_time

    - get_job_start_time:
        assign:
          - importJobStartTime: ${sys.now()}
        next: import_job_startup_wait

    - import_job_startup_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: initial_get_import_job

    - initial_get_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_status_available

    - check_import_job_status_available:
        switch:
          - condition: ${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
            next: kill_import_job
          - condition: ${"status" in IMPORT_JOB_STATUS.body}
            next: check_import_job_done
        next: import_job_status_wait

    - import_job_status_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_import_job_status_available

    - check_import_job_done:
        switch:
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
            next: the_end
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
            raise: ${IMPORT_JOB_STATUS}
          - condition: ${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
            next: kill_import_job
        next: import_job_wait

    - get_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_done

    - import_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: get_import_job

    - kill_import_job:
        call: http.post
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: get_killed_import_job

    - get_killed_import_job:
        call: http.get
        args:
          url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: KILLED_IMPORT_JOB_STATUS
        next: killed

    - killed:
        raise: ${KILLED_IMPORT_JOB_STATUS}

    - the_end:
        return: ${IMPORT_JOB_STATUS}

定义 Bash 变量，创建工作流，并酌情创建运行时间表以运行流水线：
```
# Define Bash variables (replace with your actual values)
project_id="PROJECT_ID"
region="LOCATION_ID"
service_account="SERVICE_ACCOUNT_ID"
workflow_source="WORKFLOW_DEFINITION_FILE.yaml"
workflow_name="WORKFLOW_NAME"
workflow_args='WORKFLOW_ARGUMENTS'

# Create Workflows resource
gcloud workflows deploy ${workflow_name} \
  --project=${project_id} \
  --location=${region} \
  --source=${workflow_source} \
  --service-account=${service_account}

# Create Cloud Scheduler job
gcloud scheduler jobs create http ${workflow_name}-scheduler \
  --project=${project_id} \
  --location=${region} \
  --schedule="CRON_SCHEDULE_EXPRESSION" \
  --time-zone="UTC" \
  --uri="https://workflowexecutions.googleapis.com/v1/projects/${project_id}/locations/${region}/workflows/${workflow_name}/executions" \
  --http-method="POST" \
  --oauth-service-account-email=${service_account} \
  --headers="Content-Type=application/json" \
  --message-body='{"argument": ${workflow_args}}'
```
替换以下内容：
- PROJECT_ID：用于导入元数据的目标 Google Cloud项目名称。
- LOCATION_ID：Dataproc Serverless 和元数据导入作业的运行位置，也是元数据的导入目标 Google Cloud 位置。
- SERVICE_ACCOUNT_ID：您在本文档的所需角色部分中配置的服务账号。
- WORKFLOW_DEFINITION_FILE：工作流定义 YAML 文件的路径。
- WORKFLOW_NAME：工作流的名称。
- WORKFLOW_ARGUMENTS：要传递给连接器的运行时参数。参数采用 JSON 格式：
```
{
    "TARGET_PROJECT_ID": "PROJECT_ID",
    "CLOUD_REGION": "LOCATION_ID",
    "TARGET_ENTRY_GROUP_ID": "ENTRY_GROUP_ID",
    "CREATE_TARGET_ENTRY_GROUP": CREATE_ENTRY_GROUP_BOOLEAN,
    "CLOUD_STORAGE_BUCKET_ID": "BUCKET_ID",
    "SERVICE_ACCOUNT": "SERVICE_ACCOUNT_ID",
    "ADDITIONAL_CONNECTOR_ARGS": [ADDITIONAL_CONNECTOR_ARGUMENTS],
    "CUSTOM_CONTAINER_IMAGE": "CONTAINER_IMAGE",
    "IMPORT_JOB_SCOPE_ENTRY_TYPES": [ENTRY_TYPES],
    "IMPORT_JOB_SCOPE_ASPECT_TYPES": [ASPECT_TYPES],
    "IMPORT_JOB_LOG_LEVEL": "INFO",
    "NETWORK_TAGS": [],
    "NETWORK_URI": "",
    "SUBNETWORK_URI": ""
 }
```
  对于 Cloud Scheduler，带英文引号的字符串中的英文双引号使用反斜杠 (\) 进行转义。例如：--message-body="{\"argument\": \"{\\\"key\\\": \\\"value\\\"}\"}"。
  
  替换以下内容：
  - ENTRY_GROUP_ID：要导入元数据的条目组 ID。条目组 ID 可以包含小写字母、数字和连字符。
    
    此条目组的完整资源名称为 projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID。
  - CREATE_ENTRY_GROUP_BOOLEAN：如果希望流水线在项目中创建尚未存在的条目组，请将此值设为 true。
  - BUCKET_ID：用于存储连接器生成的元数据导入文件的 Cloud Storage 存储桶的名称。每次执行工作流都会创建一个新文件夹。
  - ADDITIONAL_CONNECTOR_ARGUMENTS：传递给连接器的附加参数列表。如需查看示例，请参阅开发用于元数据导入的自定义连接器。
  - CONTAINER_IMAGE：托管在 Artifact Registry 中的连接器自定义容器映像。
  - ENTRY_TYPES：导入范围内的条目类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
  - ASPECT_TYPES：导入范围内的方面类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
  - 可选：对于 NETWORK_TAGS 参数，提供网络标记列表。
  - 可选：对于 NETWORK_URI 参数，请提供连接到数据源的 VPC 网络的 URI。如果您提供网络，请省略子网参数。
  - 可选：对于 SUBNETWORK_URI 参数，请提供连接到数据源的子网 URI。如果您提供了子网，请省略网络参数。
- CRON_SCHEDULE_EXPRESSION：用于定义流水线运行时间表的 cron 表达式。例如，如需每天在午夜运行时间表，请使用表达式 0 0 * * *。
如需按需运行流水线，请执行工作流：
```
workflow_name="WORKFLOW_NAME"
workflow_args='WORKFLOW_ARGUMENTS'

gcloud workflows run "${workflow_name}" --project=${project_id} --location=${location} --data '${workflow_args}'
```
工作流参数采用 JSON 格式，但未进行转义。

根据导入的元数据量不同，工作流运行可能需要几分钟或更长时间。如需详细了解如何查看进度，请参阅访问工作流执行结果。

流水线运行完毕后，您可以在通用目录中搜索导入的元数据。

Terraform

克隆 cloud-dataplex代码库。

该代码库包含以下 Terraform 文件：
- main.tf：定义要创建的 Google Cloud 资源。
- variables.tf：声明变量。
- byo-connector.tfvars：定义托管式连接流水线的变量。

修改 .tfvars 文件，将占位符替换为连接器的信息。

project_id                      = "PROJECT_ID"
region                          = "LOCATION_ID"
service_account                 = "SERVICE_ACCOUNT_ID"
cron_schedule                   = "CRON_SCHEDULE_EXPRESSION"
workflow_args                   = {"TARGET_PROJECT_ID": "PROJECT_ID", "CLOUD_REGION": "LOCATION_ID", "TARGET_ENTRY_GROUP_ID": "ENTRY_GROUP_ID", "CREATE_TARGET_ENTRY_GROUP": CREATE_ENTRY_GROUP_BOOLEAN, "CLOUD_STORAGE_BUCKET_ID": "BUCKET_ID", "SERVICE_ACCOUNT": "SERVICE_ACCOUNT_ID", "ADDITIONAL_CONNECTOR_ARGS": [ADDITIONAL_CONNECTOR_ARGUMENTS], "CUSTOM_CONTAINER_IMAGE": "CONTAINER_IMAGE", "IMPORT_JOB_SCOPE_ENTRY_TYPES": [ENTRY_TYPES], "IMPORT_JOB_SCOPE_ASPECT_TYPES": [ASPECT_TYPES], "IMPORT_JOB_LOG_LEVEL": "INFO", "NETWORK_TAGS": [], "NETWORK_URI": "", "SUBNETWORK_URI": ""}


workflow_source                 = <<EOF
main:
  params: [args]
  steps:
    - init:
        assign:
        - WORKFLOW_ID: $${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
        - NETWORK_URI: $${default(map.get(args, "NETWORK_URI"), "")}
        - SUBNETWORK_URI: $${default(map.get(args, "SUBNETWORK_URI"), "")}
        - NETWORK_TAGS: $${default(map.get(args, "NETWORK_TAGS"), [])}

    - check_networking:
        switch:
          - condition: $${NETWORK_URI != "" and SUBNETWORK_URI != ""}
            raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
          - condition: $${NETWORK_URI != ""}
            steps:
              - submit_extract_job_with_network_uri:
                  assign:
                    - NETWORKING: $${NETWORK_URI}
                    - NETWORK_TYPE: "networkUri"
          - condition: $${SUBNETWORK_URI != ""}
            steps:
              - submit_extract_job_with_subnetwork_uri:
                  assign:
                    - NETWORKING: $${SUBNETWORK_URI}
                    - NETWORK_TYPE: "subnetworkUri"
        next: set_default_networking

    - set_default_networking:
        assign:
          - NETWORK_TYPE: "networkUri"
          - NETWORKING: $${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}
        next: check_create_target_entry_group

    - check_create_target_entry_group:
        switch:
          - condition: $${args.CREATE_TARGET_ENTRY_GROUP == true}
            next: create_target_entry_group
          - condition: $${args.CREATE_TARGET_ENTRY_GROUP == false}
            next: generate_extract_job_link

    - create_target_entry_group:
        call: http.post
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: generate_extract_job_link

    - generate_extract_job_link:
        call: sys.log
        args:
            data: $${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
            severity: "INFO"
        next: submit_pyspark_extract_job

    - submit_pyspark_extract_job:
        call: http.post
        args:
          url: $${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          headers:
            Content-Type: "application/json"
          query:
            batchId: $${WORKFLOW_ID}
          body:
            pysparkBatch:
              mainPythonFileUri: file:///main.py
              args:
                - $${"--target_project_id=" + args.TARGET_PROJECT_ID}
                - $${"--target_location_id=" + args.CLOUD_REGION}
                - $${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                - $${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                - $${"--output_folder=" + WORKFLOW_ID}
                - $${args.ADDITIONAL_CONNECTOR_ARGS}
            runtimeConfig:
                containerImage: $${args.CUSTOM_CONTAINER_IMAGE}
            environmentConfig:
                executionConfig:
                    serviceAccount: $${args.SERVICE_ACCOUNT}
                    stagingBucket: $${args.CLOUD_STORAGE_BUCKET_ID}
                    $${NETWORK_TYPE}: $${NETWORKING}
                    networkTags: $${NETWORK_TAGS}
        result: RESPONSE_MESSAGE
        next: check_pyspark_extract_job

    - check_pyspark_extract_job:
        call: http.get
        args:
          url: $${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: PYSPARK_EXTRACT_JOB_STATUS
        next: check_pyspark_extract_job_done

    - check_pyspark_extract_job_done:
        switch:
          - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
            next: generate_import_logs_link
          - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
            raise: $${PYSPARK_EXTRACT_JOB_STATUS}
          - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
            raise: $${PYSPARK_EXTRACT_JOB_STATUS}
        next: pyspark_extract_job_wait

    - pyspark_extract_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_pyspark_extract_job

    - generate_import_logs_link:
        call: sys.log
        args:
            data: $${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
            severity: "INFO"
        next: submit_import_job

    - submit_import_job:
        call: http.post
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
          body:
            type: IMPORT
            import_spec:
              source_storage_uri: $${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
              entry_sync_mode: FULL
              aspect_sync_mode: INCREMENTAL
              log_level: $${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
              scope:
                entry_groups: 
                  - $${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                entry_types: $${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                aspect_types: $${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
        result: IMPORT_JOB_RESPONSE
        next: get_job_start_time

    - get_job_start_time:
        assign:
          - importJobStartTime: $${sys.now()}
        next: import_job_startup_wait

    - import_job_startup_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: initial_get_import_job

    - initial_get_import_job:
        call: http.get
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_status_available

    - check_import_job_status_available:
        switch:
          - condition: $${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
            next: kill_import_job
          - condition: $${"status" in IMPORT_JOB_STATUS.body}
            next: check_import_job_done
        next: import_job_status_wait

    - import_job_status_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: check_import_job_status_available

    - check_import_job_done:
        switch:
          - condition: $${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
            next: the_end
          - condition: $${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
            raise: $${IMPORT_JOB_STATUS}
          - condition: $${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
            raise: $${IMPORT_JOB_STATUS}
          - condition: $${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
            raise: $${IMPORT_JOB_STATUS}
          - condition: $${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
            next: kill_import_job
        next: import_job_wait

    - get_import_job:
        call: http.get
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: IMPORT_JOB_STATUS
        next: check_import_job_done

    - import_job_wait:
        call: sys.sleep
        args:
          seconds: 30
        next: get_import_job

    - kill_import_job:
        call: http.post
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        next: get_killed_import_job

    - get_killed_import_job:
        call: http.get
        args:
          url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
          auth:
            type: OAuth2
            scopes: "https://www.googleapis.com/auth/cloud-platform"
        result: KILLED_IMPORT_JOB_STATUS
        next: killed

    - killed:
        raise: $${KILLED_IMPORT_JOB_STATUS}

    - the_end:
        return: $${IMPORT_JOB_STATUS}
EOF

替换以下内容：

PROJECT_ID：用于导入元数据的目标 Google Cloud项目名称。
LOCATION_ID：Dataproc Serverless 和元数据导入作业的运行位置，也是元数据的导入目标 Google Cloud 位置。
SERVICE_ACCOUNT_ID：您在本文档的所需角色部分中配置的服务账号。
CRON_SCHEDULE_EXPRESSION：用于定义流水线运行时间表的 cron 表达式。例如，如需每天在午夜运行时间表，请使用表达式 0 0 * * *。
ENTRY_GROUP_ID：要导入元数据的条目组 ID。条目组 ID 可以包含小写字母、数字和连字符。

此条目组的完整资源名称为 projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID。
CREATE_ENTRY_GROUP_BOOLEAN：如果希望流水线在项目中创建尚未存在的条目组，请将此值设为 true。
BUCKET_ID：用于存储连接器生成的元数据导入文件的 Cloud Storage 存储桶的名称。每次执行工作流都会创建一个新文件夹。
ADDITIONAL_CONNECTOR_ARGUMENTS：传递给连接器的附加参数列表。如需查看示例，请参阅开发用于元数据导入的自定义连接器。请使用双引号将每个参数括起来，并用英文逗号分隔各个参数。
CONTAINER_IMAGE：托管在 Artifact Registry 中的连接器自定义容器映像。
ENTRY_TYPES：导入范围内的条目类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
ASPECT_TYPES：导入范围内的方面类型列表，格式为 projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID。LOCATION_ID 必须与您导入元数据的Google Cloud 位置相同，或者是 global。
可选：对于 NETWORK_TAGS 参数，提供网络标记列表。
可选：对于 NETWORK_URI 参数，请提供连接到数据源的 VPC 网络的 URI。如果您提供网络，请省略子网参数。
可选：对于 SUBNETWORK_URI 参数，请提供连接到数据源的子网 URI。如果您提供了子网，请省略网络参数。

初始化 Terraform：
```
terraform init
```
使用 .tfvars 文件验证 Terraform：
```
terraform plan --var-file=CONNECTOR_VARIABLES_FILE.tfvars
```
将 CONNECTOR_VARIABLES_FILE 替换为变量定义文件的名称。
使用 .tfvars 文件部署 Terraform：
```
terraform apply --var-file=CONNECTOR_VARIABLES_FILE.tfvars
```
Terraform 会在指定项目中创建工作流和 Cloud Scheduler 作业。Workflows 会按照您指定的时间表运行流水线。

根据导入的元数据量不同，工作流运行可能需要几分钟或更长时间。如需详细了解如何查看进度，请参阅访问工作流执行结果。

流水线运行完毕后，您可以在通用目录中搜索导入的元数据。

查看作业日志

使用 Cloud Logging 查看托管式连接流水线的日志。日志载荷包含指向 Dataproc Serverless 批量作业和元数据导入作业的日志链接（如适用）。如需了解详情，请参阅查看工作流日志。

问题排查

请按照以下问题排查建议操作：

为元数据作业配置导入作业日志级别，以使用调试级日志记录，而不是信息级日志记录。
查看 Dataproc Serverless 批量作业（适用于连接器运行）和元数据导入作业的日志。如需了解详情，请参阅查询 Dataproc Serverless for Spark 日志和查询元数据作业日志。
如果某个条目无法通过流水线导入，并且错误消息未提供足够的信息，请尝试在测试条目组中创建具有相同详细信息的自定义条目。如需了解详情，请参阅创建自定义条目。

使用工作流从自定义来源导入元数据 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

准备工作

构建连接器

配置 Google Cloud 资源

所需的角色

控制台

gcloud

导入元数据

控制台

gcloud

Terraform

查看作业日志

问题排查

后续步骤

使用工作流从自定义来源导入元数据
使用集合让一切井井有条根据您的偏好保存内容并对其进行分类。