使用 Workflows 從自訂來源匯入中繼資料

Workflows 中的代管連線管道可讓您從自訂第三方資料來源擷取中繼資料，並匯入 Knowledge Catalog (原稱 Dataplex Universal Catalog)。您可以視需要或定期執行管道，讓目錄保持同步，以便探索、搜尋及管理外部資產。

如要進一步瞭解代管連線，請參閱「代管連線總覽」。

事前準備

匯入中繼資料前，請先完成本節中的工作。

建構連接器

連接器會從資料來源擷取中繼資料，並產生可由 Knowledge Catalog 匯入的中繼資料匯入檔案。連接器是 Artifact Registry 映像檔，可在 Managed Service for Apache Spark 上執行。

• 建構自訂連接器，從第三方來源擷取中繼資料。

  如需連接器範例，做為建構連接器的參考範本，請參閱「開發自訂連接器以匯入中繼資料」。

設定 Google Cloud 資源

1. 啟用 Workflows、Dataproc、Cloud Storage、Dataplex、Secret Manager、Artifact Registry 和 Cloud Scheduler API (如有尚未啟用的 API)。

   啟用 API 所需的角色

   您必須具備 serviceusage.services.enable 權限，才能啟用 API。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「Service Usage 管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

   啟用 API

   如果您不打算按照排程執行管道，則不需要啟用 Cloud Scheduler API。

2. 在 Secret Manager 中建立密鑰，儲存第三方資料來源的憑證。

3. 設定虛擬私有雲 (VPC) 網路，以便執行 Managed Service for Apache Spark 工作負載。

4. 建立 Cloud Storage bucket，用於儲存中繼資料匯入檔案。

5. 建立下列 Knowledge Catalog 資源：

   1. 為要匯入的項目建立自訂切面類型。

   2. 為要匯入的項目建立自訂項目類型。

必要的角色

服務帳戶代表工作流程的身分，並決定工作流程具備的權限，以及可存取的 Google Cloud 資源。您需要 Workflows 的服務帳戶 (用於執行管道)，以及 Managed Service for Apache Spark 的服務帳戶 (用於執行連接器)。

您可以使用 Compute Engine 預設服務帳戶 ([e-mail])，或建立自己的服務帳戶，執行代管連線管道。

控制台

1. 前往 Google Cloud 控制台的「IAM」(身分與存取權管理) 頁面。

   前往「身分與存取權管理」頁面

2. 選取要匯入中繼資料的專案。

3. 按一下「 授予存取權」，然後輸入服務帳戶的電子郵件地址。

4. 將下列角色指派給服務帳戶：

   • 記錄檔寫入者 (roles/logging.logWriter)
   • Dataplex 項目群組擁有者 (roles/dataplex.entryGroupOwner)
   • Dataplex 工作中繼資料擁有者 (roles/dataplex.metadataJobOwner)
   • Dataplex Catalog 編輯者 (roles/dataplex.catalogEditor)
   • Dataproc 編輯者 (roles/dataproc.editor)
   • Dataproc worker (roles/dataproc.worker)
   • Secret Manager 密鑰存取者 (roles/secretmanager.secretAccessor)：用於儲存資料來源憑證的密鑰
   • Storage 物件使用者 (roles/storage.objectUser) - 位於 Cloud Storage bucket
   • Artifact Registry 讀取者 (roles/artifactregistry.reader)：適用於含有連接器映像檔的 Artifact Registry 存放區
   • 服務帳戶使用者 (roles/iam.serviceAccountUser)：如果您使用不同的服務帳戶，請將這個角色授予執行 Workflows 的服務帳戶，並套用至執行 Managed Service for Apache Spark 批次工作的服務帳戶
   • Workflows 叫用者 (roles/workflows.invoker)：如要排定管道

5. 儲存變更。

gcloud

1. 將角色授予服務帳戶。執行下列指令：

   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：服務帳戶，例如 [e-mail]。

2. 在資源層級，將下列角色授予服務帳戶：

   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="serviceAccount:SERVICE_ACCOUNT_ID" \
       --role=roles/artifactregistry.reader
   

   更改下列內容：

   • SECRET_ID：儲存資料來源憑證的 Secret ID。格式為 projects/PROJECT_ID/secrets/SECRET_ID。
   • BUCKET_ID：Cloud Storage bucket 的名稱。
   • REPOSITORY：含有連接器映像檔的 Artifact Registry 存放區。
   • REPOSITORY_LOCATION：託管存放區的 Google Cloud位置。

3. 將服務帳戶使用者 (roles/iam.serviceAccountUser) 角色，授予執行 Workflows 的服務帳戶，該角色位於執行 Managed Service for Apache Spark 批次工作的服務帳戶。即使 Workflows 和 Managed Service for Apache Spark 使用相同的服務帳戶，您仍須授予這個角色。

   gcloud iam service-accounts add-iam-policy-binding \
       serviceAccount:SERVICE_ACCOUNT_ID \
       --member="serviceAccount:SERVICE_ACCOUNT_ID" \
       --role='roles/iam.serviceAccountUser'
   

   如果使用不同的服務帳戶，--member 旗標的值就是執行 Managed Service for Apache Spark 批次工作的服務帳戶。

4. 如要排定管道，請將 Workflows 叫用者 (roles/workflows.invoker) 角色授予服務帳戶：

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

匯入中繼資料

如要匯入中繼資料，請建立並執行會執行代管連線管道的工作流程。這個管道會從自訂資料來源擷取中繼資料、建立必要的項目群組，並將中繼資料匯入 Knowledge Catalog。您也可以建立執行管道的時間表。

控制台

1. 建立工作流程。用下列資訊作答：

   • 服務帳戶：您在本文件「必要角色」一節中設定的服務帳戶。

   • 加密：選取「Google-managed encryption key」。

     注意： 客戶自行管理的加密金鑰 (CMEK) 不會傳播至 Knowledge Catalog 和 Managed Service for Apache Spark 工作，以進行端對端加密。

   • 定義工作流程：提供下列定義檔：

     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: prepare_pyspark_job_body
     
         - 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: prepare_pyspark_job_body
     
         - prepare_pyspark_job_body:
             assign:
               - pyspark_batch_body:
                   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}
             next: add_jar_file_uri_if_present
     
         - add_jar_file_uri_if_present:
             switch:
               - condition: ${args.JAR_FILE_URI != "" and args.JAR_FILE_URI != null}
                 assign:
                   - pyspark_batch_body.jarFileUris : ${args.JAR_FILE_URI}
             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: ${pyspark_batch_body}
                 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}
     

2. 如要視需要執行管道，請執行工作流程。

   提供下列執行階段引數：

   {
       "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",
       "JAR_FILE_URI": "",
       "NETWORK_TAGS": [],
       "NETWORK_URI": "",
       "SUBNETWORK_URI": ""
    }
   

   更改下列內容：

   • PROJECT_ID：要匯入中繼資料的目標 Google Cloud專案名稱。
   • LOCATION_ID：Managed Service for Apache Spark 和中繼資料匯入工作的執行位置，以及中繼資料的匯入位置。 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 bucket 的名稱，用於儲存連接器產生的中繼資料匯入檔案。每次執行工作流程都會建立新資料夾。
   • SERVICE_ACCOUNT_ID：您在本文件「必要角色」一節中設定的服務帳戶。服務帳戶會在 Managed Service for Apache Spark 中執行連接器。
   • 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 引數中，提供連線至資料來源的虛擬私有雲網路 URI。如果提供網路，請省略子網路引數。
   • 選用：在 SUBNETWORK_URI 引數中，提供連至資料來源的子網路 URI。如果提供子網路，請省略網路引數。

   視匯入的中繼資料量而定，管道可能需要幾分鐘或更久的時間才能執行。如要進一步瞭解如何查看進度，請參閱「存取工作流程執行結果」。

   管道執行完畢後，您可以在 Knowledge Catalog 搜尋匯入的中繼資料。

3. 選用：如要依排程執行管道，請使用 Cloud Scheduler 建立排程。用下列資訊作答：

   • 頻率：定義管道執行排程的 Unix-cron 運算式。
   • 工作流程引數：連接器的執行階段引數，如上一個步驟所述。
   • 服務帳戶：服務帳戶。服務帳戶會管理排程器。

gcloud

1. 將下列工作負載定義儲存為 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: prepare_pyspark_job_body
   
       - 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: prepare_pyspark_job_body
   
       - prepare_pyspark_job_body:
           assign:
             - pyspark_batch_body:
                 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}
           next: add_jar_file_uri_if_present
   
       - add_jar_file_uri_if_present:
           switch:
             - condition: ${args.JAR_FILE_URI != "" and args.JAR_FILE_URI != null}
               assign:
                 - pyspark_batch_body.jarFileUris : ${args.JAR_FILE_URI}
           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: ${pyspark_batch_body}
               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}
   

2. 定義 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：Managed Service for Apache Spark 和中繼資料匯入工作的執行位置，以及中繼資料的匯入位置。 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",
         "JAR_FILE_URI": "",
         "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 bucket 的名稱，用於儲存連接器產生的中繼資料匯入檔案。每次執行工作流程都會建立新資料夾。
     • 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 引數中，提供連線至資料來源的虛擬私有雲網路 URI。如果提供網路，請省略子網路引數。
     • 選用：在 SUBNETWORK_URI 引數中，提供連至資料來源的子網路 URI。如果提供子網路，請省略網路引數。

   • CRON_SCHEDULE_EXPRESSION：定義管道執行排程的 cron 運算式。舉例來說，如要在每天午夜執行排程，請使用 0 0 * * * 運算式。

3. 如要視需要執行管道，請執行工作流程：

   workflow_name="WORKFLOW_NAME"
   workflow_args='WORKFLOW_ARGUMENTS'
   
   gcloud workflows run "${workflow_name}" --project=${project_id} --location=${location} --data '${workflow_args}'
   

   工作流程引數採用 JSON 格式，但未逸出。

   視匯入的中繼資料量而定，工作流程可能需要幾分鐘或更久的時間才能執行。如要進一步瞭解如何查看進度，請參閱「存取工作流程執行結果」。

   管道執行完畢後，您可以在 Knowledge Catalog 搜尋匯入的中繼資料。

Terraform

1. 複製 cloud-dataplex 存放區。

   存放區包含下列 Terraform 檔案：

   • main.tf：定義要建立的 Google Cloud 資源。
   • variables.tf：宣告變數。
   • byo-connector.tfvars：定義代管連線管道的變數。

2. 編輯 .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：Managed Service for Apache Spark 和中繼資料匯入工作的執行位置，以及中繼資料的匯入位置。 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 bucket 的名稱，用於儲存連接器產生的中繼資料匯入檔案。每次執行工作流程都會建立新資料夾。
   • 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 引數中，提供連線至資料來源的虛擬私有雲網路 URI。如果提供網路，請省略子網路引數。
   • 選用：在 SUBNETWORK_URI 引數中，提供連至資料來源的子網路 URI。如果提供子網路，請省略網路引數。

3. 初始化 Terraform：

   terraform init
   

4. 使用 .tfvars 檔案驗證 Terraform：

   terraform plan --var-file=CONNECTOR_VARIABLES_FILE.tfvars
   

   將 CONNECTOR_VARIABLES_FILE 替換為變數定義檔案的名稱。

5. 使用 .tfvars 檔案部署 Terraform：

   terraform apply --var-file=CONNECTOR_VARIABLES_FILE.tfvars
   

   Terraform 會在指定專案中建立工作流程和 Cloud Scheduler 工作。Workflows 會按照您指定的排程執行管道。

   視匯入的中繼資料量而定，工作流程可能需要幾分鐘或更久的時間才能執行。如要進一步瞭解如何查看進度，請參閱「存取工作流程執行結果」。

   管道執行完畢後，您可以在 Knowledge Catalog 搜尋匯入的中繼資料。

查看工作記錄檔

使用 Cloud Logging 查看代管連線管道的記錄。記錄酬載包含 Managed Service for Apache Spark 批次工作和中繼資料匯入工作的記錄連結 (如適用)。詳情請參閱「查看工作流程記錄」。

疑難排解

請嘗試下列疑難排解建議：

• 將中繼資料工作的匯入工作記錄層級，設為使用偵錯層級記錄，而非資訊層級記錄。
• 查看 Managed Service for Apache Spark 批次工作 (用於連接器執行作業) 和中繼資料匯入工作的記錄。詳情請參閱「查詢 Managed Service for Apache Spark 記錄」和「查詢中繼資料工作記錄」。
• 如果無法使用管道匯入項目，且錯誤訊息未提供足夠資訊，請嘗試在測試項目群組中，建立具有相同詳細資料的自訂項目。詳情請參閱「建立自訂項目」。

後續步驟

• 關於 Knowledge Catalog 的中繼資料管理功能
• 開發中繼資料匯入專用的自訂連接器