Workflows を使用してカスタムソースからメタデータをインポートする

このドキュメントでは、Workflows でマネージド接続パイプラインを実行して、サードパーティ ソースから Dataplex にメタデータをインポートする方法について説明します。

マネージド接続パイプラインを設定するには、データソースのコネクタを構築します。次に、Workflows でパイプラインを実行します。このパイプラインは、データソースからメタデータを抽出し、メタデータを Dataplex にインポートします。必要に応じて、パイプラインは Google Cloud プロジェクトに Dataplex Catalog エントリ グループも作成します。

マネージド接続の詳細については、マネージド接続の概要をご覧ください。

始める前に

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

コネクタの作成

コネクタはデータソースからメタデータを抽出し、Dataplex によってインポートできるメタデータ インポート ファイルを生成します。コネクタは、Dataproc Serverless で実行できる Artifact Registry イメージです。

• サードパーティ ソースからメタデータを抽出するカスタム コネクタを構築します。

  独自のコネクタを構築するための参照テンプレートとして使用できるコネクタの例については、メタデータのインポート用にカスタム コネクタを開発するをご覧ください。

Google Cloud リソースを構成する

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

   Enable the APIs

   パイプラインをスケジュールに従って実行する予定がない場合は、Cloud Scheduler API を有効にする必要はありません。

2. Secret Manager でシークレットを作成して、サードパーティのデータソースの認証情報を保存します。

3. Dataproc Serverless for Spark ワークロードを実行するように Virtual Private Cloud（VPC）ネットワークを構成する。

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

5. 次の Dataplex Catalog リソースを作成します。

   1. インポートするエントリのカスタム アスペクト タイプを作成します。

   2. インポートするエントリのカスタム エントリタイプを作成します。

必要なロール

サービス アカウントはワークフローの ID を表し、ワークフローが持つ権限と、アクセスできる Google Cloud リソースを定めます。Workflows 用（パイプラインを実行するため）と Dataproc Serverless 用（コネクタを実行するため）のサービス アカウントが必要です。

Compute Engine のデフォルトのサービス アカウント（[e-mail]）を使用するか、独自のサービス アカウント（またはアカウント）を作成して、マネージド接続パイプラインを実行できます。

Console

1. Google Cloud コンソールの [IAM] ページに移動します。

   [IAM] に移動

2. メタデータをインポートするプロジェクトを選択します。

3. [アクセス権を付与] をクリックし、サービス アカウントのメールアドレスを入力します。

4. サービス アカウントに次のロールを割り当てます。

   • ログ書き込み
   • Dataplex エントリ グループ オーナー
   • Dataplex メタデータ ジョブ オーナー
   • Dataplex Catalog 編集者
   • Dataproc 編集者
   • Dataproc ワーカー
   • Secret Manager Secret Accessor - データソースの認証情報を保存するシークレット
   • Storage オブジェクト ユーザー - Cloud Storage バケットに対して
   • Artifact Registry 読み取り - コネクタ イメージを含む Artifact Registry リポジトリ
   • サービス アカウント ユーザー - 異なるサービス アカウントを使用する場合は、ワークフローを実行するサービス アカウントに、Dataproc サーバーレス バッチジョブを実行するサービス アカウントのこのロールを付与します。
   • Workflows 起動元 - パイプラインのスケジュールを設定する必要がある場合

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

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

   • SECRET_ID: データソースの認証情報を格納するシークレットの ID。形式は projects/PROJECT_ID/secrets/SECRET_ID です。
   • BUCKET_ID: Cloud Storage バケットの名前。
   • REPOSITORY: コネクタ イメージを含む Artifact Registry リポジトリ。
   • REPOSITORY_LOCATION: リポジトリがホストされている Google Cloud のロケーション。

3. 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 サーバーレス バッチジョブを実行するサービス アカウントです。

4. パイプラインのスケジュールを設定する場合は、サービス アカウントに次のロールを付与します。

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

メタデータのインポート

メタデータをインポートするには、マネージド接続パイプラインを実行するワークフローを作成して実行します。必要に応じて、パイプラインの実行スケジュールを作成することもできます。

Console

1. ワークフローを作成する。次の情報をお知らせください。

   • サービス アカウント: このドキュメントの必要なロールで構成したサービス アカウント。

   • 暗号化: [Google が管理する暗号鍵] を選択します。

     注: エンドツーエンドの暗号化のために、顧客管理の暗号鍵（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 != ""}
                 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}

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

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

   • PROJECT_ID: メタデータをインポートするターゲットの Google Cloud プロジェクトの名前。
   • LOCATION_ID: Dataproc サーバーレス ジョブとメタデータ インポート ジョブが実行され、メタデータがインポートされるターゲットの 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 サーバーレスでコネクタを実行します。
   • 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 を指定します。サブネットを指定する場合は、network 引数を省略します。

   インポートするメタデータの量によっては、パイプラインの実行に数分以上かかることがあります。進行状況を確認する方法については、ワークフローの実行結果にアクセスするをご覧ください。

   パイプラインの実行が完了したら、Dataplex 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 != ""}
               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}

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: Dataproc サーバーレス ジョブとメタデータ インポート ジョブが実行され、メタデータがインポートされるターゲットの 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 を指定します。サブネットを指定する場合は、network 引数を省略します。

   • 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 形式ですが、エスケープされていません。

   インポートするメタデータの量によっては、ワークフローの実行に数分以上かかることがあります。進行状況を確認する方法の詳細については、ワークフローの実行結果にアクセスするをご覧ください。

   パイプラインの実行が完了したら、Dataplex 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: Dataproc サーバーレス ジョブとメタデータ インポート ジョブが実行され、メタデータがインポートされるターゲットの 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 を指定します。サブネットを指定する場合は、network 引数を省略します。

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 は、指定したスケジュールでパイプラインを実行します。

   インポートするメタデータの量によっては、ワークフローの実行に数分以上かかることがあります。進行状況を確認する方法の詳細については、ワークフローの実行結果にアクセスするをご覧ください。

   パイプラインの実行が完了したら、Dataplex Catalog でインポートされたメタデータを検索できます。

ジョブのログを表示

Cloud Logging を使用して、マネージド接続パイプラインのログを表示します。ログペイロードには、関連する場合は、Dataproc Serverless バッチジョブとメタデータ インポート ジョブのログへのリンクが含まれます。詳細については、ワークフローのログを表示するをご覧ください。

トラブルシューティング

次のトラブルシューティングのヒントを参考にしてください。

• 情報レベルのロギングではなくデバッグレベルのロギングを使用するように、メタデータ ジョブのインポート ジョブのログレベルを構成します。
• Dataproc Serverless バッチジョブ（コネクタの実行用）とメタデータ インポート ジョブのログを確認します。詳細については、Dataproc Serverless for Spark のログにクエリを実行すると メタデータ ジョブログにクエリを実行するをご覧ください。
• パイプラインを使用してエントリをインポートできず、エラー メッセージに十分な情報がない場合は、テスト エントリ グループに同じ詳細情報を持つカスタム エントリを作成してみてください。詳細については、カスタム エントリを作成するをご覧ください。

次のステップ

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