Mulai 10 April 2026, Dataplex Universal Catalog kini disebut Knowledge Catalog. Nama API, library klien, CLI, dan IAM tidak berubah. Untuk mengetahui informasi selengkapnya, lihat Memperkenalkan Katalog Pengetahuan Google Cloud.

Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Mengimpor metadata dari sumber kustom menggunakan Alur Kerja

Pipeline konektivitas terkelola di Workflows memungkinkan Anda mengekstrak dan mengimpor metadata dari sumber data pihak ketiga kustom ke Knowledge Catalog (sebelumnya Dataplex Universal Catalog). Anda dapat menjalankan pipeline sesuai permintaan atau sesuai jadwal berulang untuk menjaga katalog tetap sinkron sehingga Anda dapat menemukan, menelusuri, dan mengatur aset eksternal.

Untuk mengetahui informasi selengkapnya tentang konektivitas terkelola, lihat Ringkasan konektivitas terkelola.

Sebelum memulai

Sebelum mengimpor metadata, selesaikan tugas di bagian ini.

Membangun konektor

Konektor mengekstrak metadata dari sumber data Anda dan membuat file impor metadata yang dapat diimpor oleh Knowledge Catalog. Konektor ini adalah image Artifact Registry yang dapat dijalankan di Managed Service untuk Apache Spark.

Membuat konektor kustom yang mengekstrak metadata dari sumber pihak ketiga Anda.

Untuk contoh konektor yang dapat Anda gunakan sebagai template referensi untuk membuat konektor Anda sendiri, lihat Mengembangkan konektor kustom untuk impor metadata.

Mengonfigurasi resource Google Cloud

Aktifkan Workflows, Dataproc, Cloud Storage, Dataplex, Secret Manager, Artifact Registry, dan Cloud Scheduler API, jika belum diaktifkan.
Peran yang diperlukan untuk mengaktifkan API
Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Service Usage Admin (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran.
Aktifkan API

Jika tidak berencana menjalankan pipeline sesuai jadwal, Anda tidak perlu mengaktifkan Cloud Scheduler API.
Buat secret di Secret Manager untuk menyimpan kredensial sumber data pihak ketiga Anda.
Konfigurasikan jaringan Virtual Private Cloud (VPC) Anda untuk menjalankan workload Managed Service for Apache Spark.
Buat bucket Cloud Storage untuk menyimpan file impor metadata.
Buat resource Knowledge Catalog berikut:
1. Buat jenis aspek kustom untuk entri yang ingin Anda impor.
2. Buat jenis entri kustom untuk entri yang ingin Anda impor.

Peran yang diperlukan

Akun layanan mewakili identitas alur kerja dan menentukan izin yang dimiliki alur kerja dan resource Google Cloud yang dapat diaksesnya. Anda memerlukan akun layanan untuk Workflows (untuk menjalankan pipeline) dan untuk Managed Service for Apache Spark (untuk menjalankan konektor).

Anda dapat menggunakan akun layanan default Compute Engine (PROJECT_NUMBER-compute@developer.gserviceaccount.com), atau membuat akun layanan Anda sendiri (atau beberapa akun) untuk menjalankan pipeline konektivitas terkelola.

Konsol

Di konsol Google Cloud , buka halaman IAM.

Buka IAM
Pilih project tempat Anda ingin mengimpor metadata.
Klik Grant Access, lalu masukkan alamat email akun layanan.
Tetapkan peran berikut ke akun layanan:
- Logs Writer (roles/logging.logWriter)
- Dataplex Entry Group Owner (roles/dataplex.entryGroupOwner)
- Dataplex Metadata Job Owner (roles/dataplex.metadataJobOwner)
- Dataplex Catalog Editor (roles/dataplex.catalogEditor)
- Dataproc Editor (roles/dataproc.editor)
- Dataproc Worker (roles/dataproc.worker)
- Secret Manager Secret Accessor (roles/secretmanager.secretAccessor) - pada secret yang menyimpan kredensial untuk sumber data Anda
- Storage Object User (roles/storage.objectUser) - di bucket Cloud Storage
- Artifact Registry Reader (roles/artifactregistry.reader) - di repositori Artifact Registry yang berisi image konektor
- Service Account User (roles/iam.serviceAccountUser) - jika Anda menggunakan akun layanan yang berbeda, berikan peran ini kepada akun layanan yang menjalankan Workflows di akun layanan yang menjalankan tugas batch Managed Service untuk Apache Spark
- Workflows Invoker (roles/workflows.invoker) - jika Anda ingin menjadwalkan pipeline
Simpan perubahan.

gcloud

Memberikan peran ke akun layanan. Jalankan perintah berikut:

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

Ganti kode berikut:

PROJECT_ID: nama project Google Cloud target untuk mengimpor metadata.
SERVICE_ACCOUNT_ID: akun layanan, seperti my-service-account@my-project.iam.gserviceaccount.com.

Berikan peran berikut kepada akun layanan di level resource:

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

Ganti kode berikut:

SECRET_ID: ID secret yang menyimpan kredensial untuk sumber data Anda. Formatnya adalah projects/PROJECT_ID/secrets/SECRET_ID.
BUCKET_ID: nama bucket Cloud Storage.
REPOSITORY: repositori Artifact Registry yang berisi image konektor.
REPOSITORY_LOCATION: lokasi Google Cloud tempat repositori dihosting.

Berikan peran Service Account User (roles/iam.serviceAccountUser) kepada akun layanan yang menjalankan Workflows di akun layanan yang menjalankan tugas batch Managed Service untuk Apache Spark. Anda harus memberikan peran ini meskipun Anda menggunakan akun layanan yang sama untuk Workflows dan Managed Service untuk Apache Spark.
```
gcloud iam service-accounts add-iam-policy-binding \
    serviceAccount:SERVICE_ACCOUNT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role='roles/iam.serviceAccountUser'
```
Jika Anda menggunakan akun layanan yang berbeda, nilai untuk flag --member adalah akun layanan yang menjalankan tugas batch Managed Service untuk Apache Spark.

Jika Anda ingin menjadwalkan pipeline, berikan peran Workflows Invoker (roles/workflows.invoker) ke akun layanan:

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

Mengimpor metadata

Untuk mengimpor metadata, buat, lalu jalankan alur kerja yang menjalankan pipeline konektivitas terkelola. Pipeline ini mengekstrak metadata dari sumber data kustom Anda, membuat grup entri yang diperlukan, dan mengimpor metadata ke Knowledge Catalog. Anda juga dapat membuat jadwal untuk menjalankan pipeline.

Konsol

Buat alur kerja. Berikan informasi berikut:

Service account: akun layanan yang Anda konfigurasi di bagian Required roles dalam dokumen ini.
Encryption: pilih Google-managed encryption key.

Catatan: Kunci enkripsi yang dikelola pelanggan (CMEK) tidak diterapkan ke tugas Knowledge Catalog dan Managed Service for Apache Spark untuk enkripsi end-to-end.

Tentukan alur kerja: berikan file definisi berikut:

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}

Untuk menjalankan pipeline sesuai permintaan, jalankan alur kerja.

Berikan argumen runtime berikut:
```
{
    "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": ""
 }
```
Ganti kode berikut:
- PROJECT_ID: nama project Google Cloud target untuk mengimpor metadata.
- LOCATION_ID: lokasi target Google Cloud tempat Managed Service untuk Apache Spark dan tugas impor metadata akan dijalankan, dan metadata akan diimpor.
- ENTRY_GROUP_ID: ID grup entri yang akan diimpor metadatanya. ID grup entri dapat berisi huruf kecil, angka, dan tanda hubung.
  
  Nama resource lengkap grup entri ini adalah projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID.
- CREATE_ENTRY_GROUP_BOOLEAN: jika Anda ingin pipeline membuat grup entri jika belum ada di project Anda, tetapkan nilai ini ke true.
- BUCKET_ID: nama bucket Cloud Storage untuk menyimpan file impor metadata yang dihasilkan oleh konektor. Setiap eksekusi alur kerja akan membuat folder baru.
- SERVICE_ACCOUNT_ID: akun layanan yang Anda konfigurasi di bagian Peran yang diperlukan dalam dokumen ini. Akun layanan menjalankan konektor di Managed Service untuk Apache Spark.
- ADDITIONAL_CONNECTOR_ARGUMENTS: daftar argumen tambahan yang akan diteruskan ke konektor. Sebagai contoh, lihat Mengembangkan konektor kustom untuk impor metadata. Sertakan setiap argumen dalam tanda kutip ganda, dan pisahkan argumen dengan koma.
- CONTAINER_IMAGE: image container kustom dari konektor yang dihosting di Artifact Registry.
- ENTRY_TYPES: daftar jenis entri yang termasuk dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
- ASPECT_TYPES: daftar jenis aspek yang berada dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
- Opsional: untuk argumen NETWORK_TAGS, berikan daftar tag jaringan.
- Opsional: Untuk argumen NETWORK_URI, berikan URI jaringan VPC yang terhubung ke sumber data. Jika Anda menyediakan jaringan, hilangkan argumen subnetwork.
- Opsional: Untuk argumen SUBNETWORK_URI, berikan URI subnetwork yang terhubung ke sumber data. Jika Anda menyediakan subnet, hilangkan argumen jaringan.
Bergantung pada jumlah metadata yang Anda impor, pipeline mungkin memerlukan waktu beberapa menit atau lebih untuk dijalankan. Untuk mengetahui informasi selengkapnya tentang cara melihat progres, lihat Mengakses hasil eksekusi alur kerja.

Setelah pipeline selesai berjalan, Anda dapat menelusuri metadata yang diimpor di Knowledge Catalog.
Opsional: Jika Anda ingin menjalankan pipeline sesuai jadwal, buat jadwal menggunakan Cloud Scheduler. Berikan informasi berikut:
- Frequency: ekspresi Unix-cron yang menentukan jadwal untuk menjalankan pipeline.
- Argumen alur kerja: argumen runtime untuk konektor, seperti yang dijelaskan pada langkah sebelumnya.
- Service account: akun layanan. Akun layanan mengelola penjadwal.

gcloud

Simpan definisi workload berikut sebagai file 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}

Tentukan variabel Bash, buat alur kerja, dan opsional buat jadwal untuk menjalankan pipeline:
```
# 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}}'
```
Ganti kode berikut:
- PROJECT_ID: nama project Google Cloud target untuk mengimpor metadata.
- LOCATION_ID: lokasi target Google Cloud tempat Managed Service untuk Apache Spark dan tugas impor metadata akan dijalankan, dan metadata akan diimpor.
- SERVICE_ACCOUNT_ID: akun layanan yang Anda konfigurasi di bagian Peran yang diperlukan dalam dokumen ini.
- WORKFLOW_DEFINITION_FILE: jalur ke file YAML definisi alur kerja.
- WORKFLOW_NAME: nama alur kerja.
- WORKFLOW_ARGUMENTS: argumen runtime yang akan diteruskan ke konektor. Argumennya dalam format 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": ""
 }
```
  Untuk Cloud Scheduler, tanda kutip ganda di dalam string yang dikutip di-escape menggunakan garis miring terbalik (\). Contoh: --message-body="{\"argument\": \"{\\\"key\\\": \\\"value\\\"}\"}".
  
  Ganti kode berikut:
  - ENTRY_GROUP_ID: ID grup entri yang akan diimpor metadatanya. ID grup entri dapat berisi huruf kecil, angka, dan tanda hubung.
    
    Nama resource lengkap grup entri ini adalah projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID.
  - CREATE_ENTRY_GROUP_BOOLEAN: jika Anda ingin pipeline membuat grup entri jika belum ada di project Anda, tetapkan nilai ini ke true.
  - BUCKET_ID: nama bucket Cloud Storage untuk menyimpan file impor metadata yang dihasilkan oleh konektor. Setiap eksekusi alur kerja akan membuat folder baru.
  - ADDITIONAL_CONNECTOR_ARGUMENTS: daftar argumen tambahan yang akan diteruskan ke konektor. Sebagai contoh, lihat Mengembangkan konektor kustom untuk impor metadata.
  - CONTAINER_IMAGE: image container kustom dari konektor yang dihosting di Artifact Registry.
  - ENTRY_TYPES: daftar jenis entri yang termasuk dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
  - ASPECT_TYPES: daftar jenis aspek yang berada dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
  - Opsional: untuk argumen NETWORK_TAGS, berikan daftar tag jaringan.
  - Opsional: Untuk argumen NETWORK_URI, berikan URI jaringan VPC yang terhubung ke sumber data. Jika Anda menyediakan jaringan, hilangkan argumen subnetwork.
  - Opsional: Untuk argumen SUBNETWORK_URI, berikan URI subnetwork yang terhubung ke sumber data. Jika Anda menyediakan subnet, hilangkan argumen jaringan.
- CRON_SCHEDULE_EXPRESSION: ekspresi cron yang menentukan jadwal untuk menjalankan pipeline. Misalnya, untuk menjalankan jadwal pada tengah malam setiap hari, gunakan ekspresi 0 0 * * *.
Untuk menjalankan pipeline sesuai permintaan, jalankan alur kerja:
```
workflow_name="WORKFLOW_NAME"
workflow_args='WORKFLOW_ARGUMENTS'

gcloud workflows run "${workflow_name}" --project=${project_id} --location=${location} --data '${workflow_args}'
```
Argumen alur kerja dalam format JSON, tetapi tidak di-escape.

Bergantung pada jumlah metadata yang Anda impor, alur kerja mungkin memerlukan waktu beberapa menit atau lebih untuk dijalankan. Untuk mengetahui informasi selengkapnya tentang cara melihat progres, lihat Mengakses hasil eksekusi alur kerja.

Setelah pipeline selesai berjalan, Anda dapat menelusuri metadata yang diimpor di Knowledge Catalog.

Terraform

Buat clone repositori cloud-dataplex.

Repositori ini menyertakan file Terraform berikut:
- main.tf: menentukan resource Google Cloud yang akan dibuat.
- variables.tf: mendeklarasikan variabel.
- byo-connector.tfvars: menentukan variabel untuk pipeline konektivitas terkelola Anda.

Edit file .tfvars untuk mengganti placeholder dengan informasi untuk konektor Anda.

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

Ganti kode berikut:

PROJECT_ID: nama project Google Cloud target untuk mengimpor metadata.
LOCATION_ID: lokasi target Google Cloud tempat Managed Service untuk Apache Spark dan tugas impor metadata akan dijalankan, dan metadata akan diimpor.
SERVICE_ACCOUNT_ID: akun layanan yang Anda konfigurasi di bagian Peran yang diperlukan dalam dokumen ini.
CRON_SCHEDULE_EXPRESSION: ekspresi cron yang menentukan jadwal untuk menjalankan pipeline. Misalnya, untuk menjalankan jadwal pada tengah malam setiap hari, gunakan ekspresi 0 0 * * *.
ENTRY_GROUP_ID: ID grup entri yang akan diimpor metadatanya. ID grup entri dapat berisi huruf kecil, angka, dan tanda hubung.

Nama resource lengkap grup entri ini adalah projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID.
CREATE_ENTRY_GROUP_BOOLEAN: jika Anda ingin pipeline membuat grup entri jika belum ada di project Anda, tetapkan nilai ini ke true.
BUCKET_ID: nama bucket Cloud Storage untuk menyimpan file impor metadata yang dihasilkan oleh konektor. Setiap eksekusi alur kerja akan membuat folder baru.
ADDITIONAL_CONNECTOR_ARGUMENTS: daftar argumen tambahan yang akan diteruskan ke konektor. Sebagai contoh, lihat Mengembangkan konektor kustom untuk impor metadata. Sertakan setiap argumen dalam tanda kutip ganda, dan pisahkan argumen dengan koma.
CONTAINER_IMAGE: image container kustom dari konektor yang dihosting di Artifact Registry.
ENTRY_TYPES: daftar jenis entri yang termasuk dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
ASPECT_TYPES: daftar jenis aspek yang berada dalam cakupan untuk diimpor, dalam format projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID. LOCATION_ID harus berupa lokasiGoogle Cloud yang sama dengan tempat Anda mengimpor metadata, atau global.
Opsional: untuk argumen NETWORK_TAGS, berikan daftar tag jaringan.
Opsional: Untuk argumen NETWORK_URI, berikan URI jaringan VPC yang terhubung ke sumber data. Jika Anda menyediakan jaringan, hilangkan argumen subnetwork.
Opsional: Untuk argumen SUBNETWORK_URI, berikan URI subnetwork yang terhubung ke sumber data. Jika Anda menyediakan subnet, hilangkan argumen jaringan.

Lakukan inisialisasi Terraform:
```
terraform init
```
Validasi Terraform dengan file .tfvars Anda:
```
terraform plan --var-file=CONNECTOR_VARIABLES_FILE.tfvars
```
Ganti CONNECTOR_VARIABLES_FILE dengan nama file definisi variabel Anda.
Deploy Terraform dengan file .tfvars Anda:
```
terraform apply --var-file=CONNECTOR_VARIABLES_FILE.tfvars
```
Terraform membuat alur kerja dan tugas Cloud Scheduler di project yang ditentukan. Workflows menjalankan pipeline sesuai jadwal yang Anda tentukan.

Bergantung pada jumlah metadata yang Anda impor, alur kerja mungkin memerlukan waktu beberapa menit atau lebih untuk dijalankan. Untuk mengetahui informasi selengkapnya tentang cara melihat progres, lihat Mengakses hasil eksekusi alur kerja.

Setelah pipeline selesai berjalan, Anda dapat menelusuri metadata yang diimpor di Knowledge Catalog.

Lihat log tugas

Gunakan Cloud Logging untuk melihat log pipeline konektivitas terkelola. Payload log mencakup link ke log untuk tugas batch Managed Service untuk Apache Spark dan tugas impor metadata, jika relevan. Untuk mengetahui informasi selengkapnya, lihat Melihat log alur kerja.

Pemecahan masalah

Gunakan saran pemecahan masalah berikut:

Konfigurasikan tingkat log tugas impor untuk tugas metadata agar menggunakan logging tingkat debug, bukan logging tingkat info.
Tinjau log untuk tugas batch Managed Service untuk Apache Spark (untuk operasi konektor) dan tugas impor metadata. Untuk mengetahui informasi selengkapnya, lihat Membuat kueri log Managed Service untuk Apache Spark dan Membuat kueri log tugas metadata.
Jika entri tidak dapat diimpor menggunakan pipeline dan pesan error tidak memberikan informasi yang cukup, coba buat entri kustom dengan detail yang sama, dalam grup entri pengujian. Untuk mengetahui informasi selengkapnya, lihat Membuat entri kustom.

Mengimpor metadata dari sumber kustom menggunakan Alur Kerja Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.

Sebelum memulai

Membangun konektor

Mengonfigurasi resource Google Cloud

Peran yang diperlukan

Konsol

gcloud

Mengimpor metadata

Konsol

gcloud

Terraform

Lihat log tugas

Pemecahan masalah

Langkah berikutnya

Mengimpor metadata dari sumber kustom menggunakan Alur Kerja