ייבוא מטא-נתונים ממקור מותאם אישית באמצעות Workflows

נכנסים לדף IAM במסוף Google Cloud .

כניסה לדף IAM

בוחרים את הפרויקט שאליו רוצים לייבא את המטא-נתונים.

לוחצים על person_add Grant Access (הענקת גישה) ומזינים את כתובת האימייל של חשבון השירות.

מקצים לחשבון השירות את התפקידים הבאים:

• Logs Writer
• Dataplex Entry Group Owner
• הבעלים של משימת מטא-נתונים ב-Dataplex
• Dataplex Catalog Editor
• עורך Dataproc
• Dataproc Worker
• ‫Secret Manager Secret Accessor – בסוד שבו מאוחסנים פרטי הכניסה למקור הנתונים
• Storage Object User – בקטגוריה של Cloud Storage
• ‫Artifact Registry Reader – במאגר Artifact Registry שמכיל את תמונת המחבר
• משתמש בחשבון שירות – אם משתמשים בחשבונות שירות שונים, צריך להקצות לחשבון השירות שמריץ את Workflows את התפקיד הזה בחשבון השירות שמריץ את משימות האצווה של Managed Service for Apache Spark
• ‫Workflows Invoker – אם רוצים לתזמן את הפייפליין

שומרים את השינויים.

נותנים לחשבון השירות תפקידים. מריצים את הפקודות הבאות:

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

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: השם של פרויקט היעד Google Cloudשאליו רוצים לייבא את המטא-נתונים.
• ‫SERVICE_ACCOUNT_ID: חשבון השירות, למשל my-service-account@my-project.iam.gserviceaccount.com.

מקצים לחשבון השירות את התפקידים הבאים ברמת המשאב:

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

מחליפים את מה שכתוב בשדות הבאים:

• ‫SECRET_ID: המזהה של הסוד שבו מאוחסנים פרטי הכניסה למקור הנתונים. הפורמט הוא projects/PROJECT_ID/secrets/SECRET_ID.
• ‫BUCKET_ID: שם הקטגוריה ב-Cloud Storage.
• ‫REPOSITORY: מאגר Artifact Registry שמכיל את תמונת המחבר.
• ‫REPOSITORY_LOCATION: המיקום Google Cloudשבו מתארח המאגר.

מקצים לחשבון השירות שמריץ את Workflows את התפקיד roles/iam.serviceAccountUser בחשבון השירות שמריץ את משימות האצווה של Managed Service for Apache Spark. צריך להעניק את התפקיד הזה גם אם משתמשים באותו חשבון שירות גם ב-Workflows וגם ב-Managed Service for Apache Spark.

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

אם משתמשים בחשבונות שירות שונים, הערך של הדגל --member הוא חשבון השירות שמריץ את משימות האצווה של Managed Service for Apache Spark.

אם רוצים לתזמן את צינור עיבוד הנתונים, צריך להקצות לחשבון השירות את התפקיד הבא:

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

יוצרים את תהליך העבודה. עליך לספק את הפרטים הבאים:

• חשבון שירות: חשבון השירות שהגדרתם בקטע תפקידים נדרשים במסמך הזה.

• הצפנה: בוחרים באפשרות Google-managed encryption key.

• הגדרת תהליך העבודה: מספקים את קובץ ההגדרה הבא:

  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: