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

במאמר הזה מוסבר איך לייבא מטא-נתונים ממקורות של צד שלישי אל Knowledge Catalog (לשעבר Dataplex Universal Catalog) באמצעות הגדרה והפעלה של צינור קישוריות מנוהל ב-Workflows. בצינור הזה, המטא-נתונים מחולצים ממקור הנתונים המותאם אישית ומייובאים אל Knowledge Catalog, וכך נוצרות קבוצות רשומות נדרשות.

מידע נוסף על קישוריות מנוהלת זמין במאמר סקירה כללית על קישוריות מנוהלת.

לפני שמתחילים

לפני שמייבאים מטא-נתונים, צריך להשלים את המשימות שבקטע הזה.

יצירת מחבר

מחבר מחלץ את המטא-נתונים ממקור הנתונים ויוצר קובץ ייבוא של מטא-נתונים שאפשר לייבא אותו אל Knowledge Catalog. המחבר הוא תמונה של Artifact Registry שאפשר להריץ ב-Managed Service for Apache Spark.

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

  דוגמה למחבר שאפשר להשתמש בו כתבנית הפניה כדי ליצור מחבר משלכם מופיעה במאמר פיתוח מחבר בהתאמה אישית לייבוא מטא-נתונים.

הגדרת Google Cloud משאבים

1. מפעילים את ממשקי ה-API של Workflows,‏ Dataproc,‏ Cloud Storage,‏ Dataplex,‏ Secret Manager,‏ Artifact Registry ו-Cloud Scheduler.

   תפקידים שנדרשים להפעלת ממשקי API

   כדי להפעיל ממשקי API, צריך את תפקיד ה-IAM 'אדמין של Service Usage' (roles/serviceusage.serviceUsageAdmin), שכולל את ההרשאה serviceusage.services.enable. איך מקצים תפקידים

   הפעלת ממשקי ה-API

   אם אתם לא מתכננים להריץ את צינור הנתונים לפי לוח זמנים, אתם לא צריכים להפעיל את Cloud Scheduler API.

2. יוצרים סודות ב-Secret Manager כדי לאחסן את פרטי הכניסה למקור הנתונים מצד שלישי.

3. הגדרת רשת של ענן וירטואלי פרטי (VPC) להרצת עומסי עבודה של Managed Service for Apache Spark.

4. יוצרים קטגוריה של Cloud Storage לאחסון קובצי ייבוא המטא-נתונים.

5. יוצרים את המשאבים הבאים ב-Knowledge Catalog:

   1. יוצרים סוגים מותאמים אישית של היבטים לרשומות שרוצים לייבא.

   2. יוצרים סוגים מותאמים אישית של רשומות לרשומות שרוצים לייבא.

התפקידים הנדרשים

חשבון שירות מייצג את הזהות של תהליך עבודה, וקובע אילו הרשאות יש לתהליך העבודה ולאילו Google Cloud משאבים הוא יכול לגשת. צריך חשבון שירות בשביל Workflows (כדי להריץ את צינור הנתונים) ובשביל Managed Service for Apache Spark (כדי להריץ את המחבר).

אתם יכולים להשתמש בחשבון השירות של Compute Engine שמוגדר כברירת מחדל ([e-mail]), או ליצור חשבון שירות משלכם (או חשבונות שירות משלכם) כדי להריץ את צינור הקישוריות המנוהל.

המסוף

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

   כניסה לדף IAM

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

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

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

   • 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 – אם רוצים לתזמן את הפייפליין

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

3. מקצים לחשבון השירות שמריץ את 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.

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

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

ייבוא מטא-נתונים

כדי לייבא מטא-נתונים, יוצרים ואז מפעילים תהליך עבודה שמריץ את צינור הקישוריות המנוהל. אפשר גם ליצור לוח זמנים להרצת צינור הנתונים.

המסוף

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

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

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

     הערה: מפתחות הצפנה בניהול הלקוח (CMEK) לא מועברים אל Knowledge Catalog ואל משימות של Managed Service for Apache Spark לצורך הצפנה מקצה לקצה.

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

     main:
       params: [args]
       steps:
         - init:
             assign:
             - WORKFLOW_ID: ${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
             - NETWORK_URI: ${default(map.get(args, "NETWORK_URI"), "")}
             - SUBNETWORK_URI: ${default(map.get(args, "SUBNETWORK_URI"), "")}
             - NETWORK_TAGS: ${default(map.get(args, "NETWORK_TAGS"), [])}
     
         - check_networking:
             switch:
               - condition: ${NETWORK_URI != "" and SUBNETWORK_URI != ""}
                 raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
               - condition: ${NETWORK_URI == "" and SUBNETWORK_URI == ""}
                 steps:
                  - submit_extract_job_with_default_network_uri:
                       assign:
                         - NETWORK_TYPE: "networkUri"
                         - NETWORKING: ${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}  
               - condition: ${NETWORK_URI != ""}
                 steps:
                   - submit_extract_job_with_network_uri:
                       assign:
                         - NETWORKING: ${NETWORK_URI}
                         - NETWORK_TYPE: "networkUri"
               - condition: ${SUBNETWORK_URI != ""}
                 steps:
                   - submit_extract_job_with_subnetwork_uri:
                       assign:
                         - NETWORKING: ${SUBNETWORK_URI}
                         - NETWORK_TYPE: "subnetworkUri"
             next: check_create_target_entry_group
     
         - check_create_target_entry_group:
             switch:
               - condition: ${args.CREATE_TARGET_ENTRY_GROUP == true}
                 next: create_target_entry_group
               - condition: ${args.CREATE_TARGET_ENTRY_GROUP == false}
                 next: prepare_pyspark_job_body
     
         - create_target_entry_group:
             call: http.post
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             next: prepare_pyspark_job_body
     
         - prepare_pyspark_job_body:
             assign:
               - pyspark_batch_body:
                   mainPythonFileUri: file:///main.py
                   args:
                     - ${"--target_project_id=" + args.TARGET_PROJECT_ID}
                     - ${"--target_location_id=" + args.CLOUD_REGION}
                     - ${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                     - ${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                     - ${"--output_folder=" + WORKFLOW_ID}
                     - ${args.ADDITIONAL_CONNECTOR_ARGS}
             next: add_jar_file_uri_if_present
     
         - add_jar_file_uri_if_present:
             switch:
               - condition: ${args.JAR_FILE_URI != "" and args.JAR_FILE_URI != null}
                 assign:
                   - pyspark_batch_body.jarFileUris : ${args.JAR_FILE_URI}
             next: generate_extract_job_link
     
         - generate_extract_job_link:
             call: sys.log
             args:
                 data: ${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
                 severity: "INFO"
             next: submit_pyspark_extract_job
     
         - submit_pyspark_extract_job:
             call: http.post
             args:
               url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
               headers:
                 Content-Type: "application/json"
               query:
                 batchId: ${WORKFLOW_ID}
               body:
                 pysparkBatch: ${pyspark_batch_body}
                 runtimeConfig:
                     containerImage: ${args.CUSTOM_CONTAINER_IMAGE}
                 environmentConfig:
                     executionConfig:
                         serviceAccount: ${args.SERVICE_ACCOUNT}
                         stagingBucket: ${args.CLOUD_STORAGE_BUCKET_ID}
                         ${NETWORK_TYPE}: ${NETWORKING}
                         networkTags: ${NETWORK_TAGS}
             result: RESPONSE_MESSAGE
             next: check_pyspark_extract_job
     
         - check_pyspark_extract_job:
             call: http.get
             args:
               url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             result: PYSPARK_EXTRACT_JOB_STATUS
             next: check_pyspark_extract_job_done
     
         - check_pyspark_extract_job_done:
             switch:
               - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
                 next: generate_import_logs_link
               - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
                 raise: ${PYSPARK_EXTRACT_JOB_STATUS}
               - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
                 raise: ${PYSPARK_EXTRACT_JOB_STATUS}
             next: pyspark_extract_job_wait
     
         - pyspark_extract_job_wait:
             call: sys.sleep
             args:
               seconds: 30
             next: check_pyspark_extract_job
     
         - generate_import_logs_link:
             call: sys.log
             args:
                 data: ${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
                 severity: "INFO"
             next: submit_import_job
     
         - submit_import_job:
             call: http.post
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
               body:
                 type: IMPORT
                 import_spec:
                   source_storage_uri: ${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
                   entry_sync_mode: FULL
                   aspect_sync_mode: INCREMENTAL
                   log_level: ${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
                   scope:
                     entry_groups: 
                       - ${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                     entry_types: ${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                     aspect_types: ${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
             result: IMPORT_JOB_RESPONSE
             next: get_job_start_time
     
         - get_job_start_time:
             assign:
               - importJobStartTime: ${sys.now()}
             next: import_job_startup_wait
     
         - import_job_startup_wait:
             call: sys.sleep
             args:
               seconds: 30
             next: initial_get_import_job
     
         - initial_get_import_job:
             call: http.get
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             result: IMPORT_JOB_STATUS
             next: check_import_job_status_available
     
         - check_import_job_status_available:
             switch:
               - condition: ${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
                 next: kill_import_job
               - condition: ${"status" in IMPORT_JOB_STATUS.body}
                 next: check_import_job_done
             next: import_job_status_wait
     
         - import_job_status_wait:
             call: sys.sleep
             args:
               seconds: 30
             next: check_import_job_status_available
     
         - check_import_job_done:
             switch:
               - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
                 next: the_end
               - condition: ${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
                 raise: ${IMPORT_JOB_STATUS}
               - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
                 raise: ${IMPORT_JOB_STATUS}
               - condition: ${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
                 raise: ${IMPORT_JOB_STATUS}
               - condition: ${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
                 next: kill_import_job
             next: import_job_wait
     
         - get_import_job:
             call: http.get
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             result: IMPORT_JOB_STATUS
             next: check_import_job_done
     
         - import_job_wait:
             call: sys.sleep
             args:
               seconds: 30
             next: get_import_job
     
         - kill_import_job:
             call: http.post
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             next: get_killed_import_job
     
         - get_killed_import_job:
             call: http.get
             args:
               url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
               auth:
                 type: OAuth2
                 scopes: "https://www.googleapis.com/auth/cloud-platform"
             result: KILLED_IMPORT_JOB_STATUS
             next: killed
     
         - killed:
             raise: ${KILLED_IMPORT_JOB_STATUS}
     
         - the_end:
             return: ${IMPORT_JOB_STATUS}
     

2. כדי להריץ את צינור הנתונים לפי דרישה, מריצים את תהליך העבודה.

   מציינים את הארגומנטים הבאים של זמן הריצה:

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

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

   • ‫PROJECT_ID: השם של פרויקט היעד Google Cloudשאליו רוצים לייבא את המטא-נתונים.
   • ‫LOCATION_ID: מיקום היעד שבו יפעלו המשימות של Managed Service for Apache Spark ושל ייבוא המטא-נתונים, ושאליו יובאו המטא-נתונים. Google Cloud
   • ‫ENTRY_GROUP_ID: המזהה של קבוצת הרשומות שאליה רוצים לייבא את המטא-נתונים. מזהה קבוצת הרשומות יכול להכיל אותיות קטנות, מספרים ומקפים.

     השם המלא של המשאב של קבוצת הרשומות הזו הוא projects/PROJECT_ID/locations/LOCATION_ID/entryGroups/ENTRY_GROUP_ID.

   • ‫CREATE_ENTRY_GROUP_BOOLEAN: אם רוצים שצינור העיבוד ייצור את קבוצת הרשומות אם היא עדיין לא קיימת בפרויקט, צריך להגדיר את הערך הזה ל-true.
   • ‫BUCKET_ID: השם של קטגוריית Cloud Storage לאחסון קובץ ייבוא המטא-נתונים שנוצר על ידי המחבר. כל הרצה של תהליך עבודה יוצרת תיקייה חדשה.
   • ‫SERVICE_ACCOUNT_ID: חשבון השירות שהגדרתם בקטע תפקידים נדרשים במסמך הזה. חשבון השירות מפעיל את המחבר ב-Managed Service for Apache Spark.
   • ‫ADDITIONAL_CONNECTOR_ARGUMENTS: רשימה של ארגומנטים נוספים להעברה למחבר. דוגמאות מופיעות במאמר פיתוח מחבר בהתאמה אישית לייבוא מטא-נתונים. מקיפים כל ארגומנט במירכאות כפולות ומפרידים בין הארגומנטים באמצעות פסיקים.
   • ‫CONTAINER_IMAGE: תמונת הקונטיינר המותאמת אישית של המחבר שמתארחת ב-Artifact Registry.
   • ‫ENTRY_TYPES: רשימה של סוגי רשומות שנכללות בהיקף של הייבוא, בפורמט projects/PROJECT_ID/locations/LOCATION_ID/entryTypes/ENTRY_TYPE_ID. ה-LOCATION_ID צריך להיות זהה למיקום שאליו מייבאים את המטא-נתונים, או global.Google Cloud
   • ‫ASPECT_TYPES: רשימה של סוגי היבטים שנכללים בייבוא, בפורמט projects/PROJECT_ID/locations/LOCATION_ID/aspectTypes/ASPECT_TYPE_ID. המיקום LOCATION_ID צריך להיות זהה למיקוםGoogle Cloud שאליו מייבאים את המטא-נתונים, או global.
   • אופציונלי: עבור הארגומנט NETWORK_TAGS, מספקים רשימה של תגי רשת.
   • אופציונלי: בארגומנט NETWORK_URI, מציינים את ה-URI של רשת ה-VPC שמקושרת למקור הנתונים. אם מציינים רשת, לא צריך לציין את הארגומנט subnetwork.
   • אופציונלי: בארגומנט SUBNETWORK_URI מציינים את ה-URI של רשת המשנה שמקושרת למקור הנתונים. אם מציינים רשת משנה, משמיטים את ארגומנט הרשת.

   בהתאם לכמות המטא-נתונים שמייבאים, יכול להיות שייקח כמה דקות או יותר להפעיל את צינור הנתונים. מידע נוסף על הצגת ההתקדמות זמין במאמר בנושא גישה לתוצאות של הרצת תהליך עבודה.

   אחרי שהפייפליין סיים לפעול, אפשר לחפש את המטא-נתונים המיובאים ב-Knowledge Catalog.

3. אופציונלי: אם רוצים להריץ את צינור הנתונים לפי לוח זמנים, יוצרים לוח זמנים באמצעות Cloud Scheduler. עליך לספק את הפרטים הבאים:

   • תדירות: ביטוי cron של Unix שמגדיר את לוח הזמנים להרצת צינור הנתונים.
   • ארגומנט של תהליך העבודה: ארגומנטי זמן הריצה של המחבר, כפי שמתואר בשלב הקודם.
   • חשבון שירות: חשבון השירות. חשבון השירות מנהל את הכלי לתזמון.

gcloud

1. שומרים את הגדרת עומס העבודה הבאה כקובץ YAML:

   main:
     params: [args]
     steps:
       - init:
           assign:
           - WORKFLOW_ID: ${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
           - NETWORK_URI: ${default(map.get(args, "NETWORK_URI"), "")}
           - SUBNETWORK_URI: ${default(map.get(args, "SUBNETWORK_URI"), "")}
           - NETWORK_TAGS: ${default(map.get(args, "NETWORK_TAGS"), [])}
   
       - check_networking:
           switch:
             - condition: ${NETWORK_URI != "" and SUBNETWORK_URI != ""}
               raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
             - condition: ${NETWORK_URI == "" and SUBNETWORK_URI == ""}
               steps:
                - submit_extract_job_with_default_network_uri:
                     assign:
                       - NETWORK_TYPE: "networkUri"
                       - NETWORKING: ${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}  
             - condition: ${NETWORK_URI != ""}
               steps:
                 - submit_extract_job_with_network_uri:
                     assign:
                       - NETWORKING: ${NETWORK_URI}
                       - NETWORK_TYPE: "networkUri"
             - condition: ${SUBNETWORK_URI != ""}
               steps:
                 - submit_extract_job_with_subnetwork_uri:
                     assign:
                       - NETWORKING: ${SUBNETWORK_URI}
                       - NETWORK_TYPE: "subnetworkUri"
           next: check_create_target_entry_group
   
       - check_create_target_entry_group:
           switch:
             - condition: ${args.CREATE_TARGET_ENTRY_GROUP == true}
               next: create_target_entry_group
             - condition: ${args.CREATE_TARGET_ENTRY_GROUP == false}
               next: prepare_pyspark_job_body
   
       - create_target_entry_group:
           call: http.post
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           next: prepare_pyspark_job_body
   
       - prepare_pyspark_job_body:
           assign:
             - pyspark_batch_body:
                 mainPythonFileUri: file:///main.py
                 args:
                   - ${"--target_project_id=" + args.TARGET_PROJECT_ID}
                   - ${"--target_location_id=" + args.CLOUD_REGION}
                   - ${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                   - ${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                   - ${"--output_folder=" + WORKFLOW_ID}
                   - ${args.ADDITIONAL_CONNECTOR_ARGS}
           next: add_jar_file_uri_if_present
   
       - add_jar_file_uri_if_present:
           switch:
             - condition: ${args.JAR_FILE_URI != "" and args.JAR_FILE_URI != null}
               assign:
                 - pyspark_batch_body.jarFileUris : ${args.JAR_FILE_URI}
           next: generate_extract_job_link
   
       - generate_extract_job_link:
           call: sys.log
           args:
               data: ${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
               severity: "INFO"
           next: submit_pyspark_extract_job
   
       - submit_pyspark_extract_job:
           call: http.post
           args:
             url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
             headers:
               Content-Type: "application/json"
             query:
               batchId: ${WORKFLOW_ID}
             body:
               pysparkBatch: ${pyspark_batch_body}
               runtimeConfig:
                   containerImage: ${args.CUSTOM_CONTAINER_IMAGE}
               environmentConfig:
                   executionConfig:
                       serviceAccount: ${args.SERVICE_ACCOUNT}
                       stagingBucket: ${args.CLOUD_STORAGE_BUCKET_ID}
                       ${NETWORK_TYPE}: ${NETWORKING}
                       networkTags: ${NETWORK_TAGS}
           result: RESPONSE_MESSAGE
           next: check_pyspark_extract_job
   
       - check_pyspark_extract_job:
           call: http.get
           args:
             url: ${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: PYSPARK_EXTRACT_JOB_STATUS
           next: check_pyspark_extract_job_done
   
       - check_pyspark_extract_job_done:
           switch:
             - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
               next: generate_import_logs_link
             - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
               raise: ${PYSPARK_EXTRACT_JOB_STATUS}
             - condition: ${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
               raise: ${PYSPARK_EXTRACT_JOB_STATUS}
           next: pyspark_extract_job_wait
   
       - pyspark_extract_job_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: check_pyspark_extract_job
   
       - generate_import_logs_link:
           call: sys.log
           args:
               data: ${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
               severity: "INFO"
           next: submit_import_job
   
       - submit_import_job:
           call: http.post
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
             body:
               type: IMPORT
               import_spec:
                 source_storage_uri: ${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
                 entry_sync_mode: FULL
                 aspect_sync_mode: INCREMENTAL
                 log_level: ${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
                 scope:
                   entry_groups: 
                     - ${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                   entry_types: ${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                   aspect_types: ${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
           result: IMPORT_JOB_RESPONSE
           next: get_job_start_time
   
       - get_job_start_time:
           assign:
             - importJobStartTime: ${sys.now()}
           next: import_job_startup_wait
   
       - import_job_startup_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: initial_get_import_job
   
       - initial_get_import_job:
           call: http.get
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: IMPORT_JOB_STATUS
           next: check_import_job_status_available
   
       - check_import_job_status_available:
           switch:
             - condition: ${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
               next: kill_import_job
             - condition: ${"status" in IMPORT_JOB_STATUS.body}
               next: check_import_job_done
           next: import_job_status_wait
   
       - import_job_status_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: check_import_job_status_available
   
       - check_import_job_done:
           switch:
             - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
               next: the_end
             - condition: ${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
               raise: ${IMPORT_JOB_STATUS}
             - condition: ${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
               raise: ${IMPORT_JOB_STATUS}
             - condition: ${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
               raise: ${IMPORT_JOB_STATUS}
             - condition: ${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
               next: kill_import_job
           next: import_job_wait
   
       - get_import_job:
           call: http.get
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: IMPORT_JOB_STATUS
           next: check_import_job_done
   
       - import_job_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: get_import_job
   
       - kill_import_job:
           call: http.post
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           next: get_killed_import_job
   
       - get_killed_import_job:
           call: http.get
           args:
             url: ${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: KILLED_IMPORT_JOB_STATUS
           next: killed
   
       - killed:
           raise: ${KILLED_IMPORT_JOB_STATUS}
   
       - the_end:
           return: ${IMPORT_JOB_STATUS}
   

2. מגדירים משתני Bash, יוצרים את תהליך העבודה, ויוצרים לוח זמנים להרצת צינור עיבוד הנתונים (אופציונלי):

   # Define Bash variables (replace with your actual values)
   project_id="PROJECT_ID"
   region="LOCATION_ID"
   service_account="SERVICE_ACCOUNT_ID"
   workflow_source="WORKFLOW_DEFINITION_FILE.yaml"
   workflow_name="WORKFLOW_NAME"
   workflow_args='WORKFLOW_ARGUMENTS'
   
   # Create Workflows resource
   gcloud workflows deploy ${workflow_name} \
     --project=${project_id} \
     --location=${region} \
     --source=${workflow_source} \
     --service-account=${service_account}
   
   # Create Cloud Scheduler job
   gcloud scheduler jobs create http ${workflow_name}-scheduler \
     --project=${project_id} \
     --location=${region} \
     --schedule="CRON_SCHEDULE_EXPRESSION" \
     --time-zone="UTC" \
     --uri="https://workflowexecutions.googleapis.com/v1/projects/${project_id}/locations/${region}/workflows/${workflow_name}/executions" \
     --http-method="POST" \
     --oauth-service-account-email=${service_account} \
     --headers="Content-Type=application/json" \
     --message-body='{"argument": ${workflow_args}}'
   

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

   • ‫PROJECT_ID: השם של פרויקט היעד Google Cloudשאליו רוצים לייבא את המטא-נתונים.
   • ‫LOCATION_ID: מיקום היעד Google Cloud שבו יפעלו המשימות של Managed Service for Apache Spark ושל ייבוא המטא-נתונים, ושאליו יובאו המטא-נתונים.
   • ‫SERVICE_ACCOUNT_ID: חשבון השירות שהגדרתם בקטע תפקידים נדרשים במסמך הזה.
   • ‫WORKFLOW_DEFINITION_FILE: הנתיב לקובץ ה-YAML של הגדרת זרימת העבודה.
   • ‫WORKFLOW_NAME: השם של תהליך העבודה.

   • ‫WORKFLOW_ARGUMENTS: הארגומנטים של זמן הריצה שמועברים למחבר. הארגומנטים הם בפורמט JSON:

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

     ב-Cloud Scheduler, המירכאות הכפולות בתוך המחרוזת המצוטטת מסומנות בתו בריחה (escape) באמצעות קו נטוי הפוך (\\). לדוגמה: --message-body="{\"argument\": \"{\\\"key\\\": \\\"value\\\"}\"}".

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

     • ‫ENTRY_GROUP_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, מציינים את ה-URI של רשת ה-VPC שמקושרת למקור הנתונים. אם מציינים רשת, לא צריך לציין את הארגומנט subnetwork.
     • אופציונלי: בארגומנט SUBNETWORK_URI מציינים את ה-URI של רשת המשנה שמקושרת למקור הנתונים. אם מציינים רשת משנה, משמיטים את ארגומנט הרשת.

   • ‫CRON_SCHEDULE_EXPRESSION: ביטוי cron שמגדיר את לוח הזמנים להרצת צינור העיבוד. לדוגמה, כדי להפעיל את התזמון בחצות בכל יום, משתמשים בביטוי 0 0 * * *.

3. כדי להריץ את צינור עיבוד הנתונים לפי דרישה, מריצים את תהליך העבודה:

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

   הארגומנטים של תהליך העבודה הם בפורמט JSON, אבל לא נעשה בהם escape.

   בהתאם לכמות המטא-נתונים שמייבאים, יכול להיות שתהליך העבודה יימשך כמה דקות או יותר. מידע נוסף על הצגת ההתקדמות זמין במאמר גישה לתוצאות של הרצת תהליך עבודה.

   אחרי שהפייפליין סיים לפעול, אפשר לחפש את המטא-נתונים המיובאים ב-Knowledge Catalog.

Terraform

1. משכפלים את מאגר cloud-dataplex.

   המאגר כולל את קובצי Terraform הבאים:

   • ‫main.tf: הגדרת המשאבים שרוצים ליצור. Google Cloud
   • ‫variables.tf: מגדיר את המשתנים.
   • ‫byo-connector.tfvars: הגדרת המשתנים של צינור הקישוריות המנוהל.

2. עורכים את הקובץ .tfvars כדי להחליף את הפלייסהולדרים בפרטים של המחבר.

   project_id                      = "PROJECT_ID"
   region                          = "LOCATION_ID"
   service_account                 = "SERVICE_ACCOUNT_ID"
   cron_schedule                   = "CRON_SCHEDULE_EXPRESSION"
   workflow_args                   = {"TARGET_PROJECT_ID": "PROJECT_ID", "CLOUD_REGION": "LOCATION_ID", "TARGET_ENTRY_GROUP_ID": "ENTRY_GROUP_ID", "CREATE_TARGET_ENTRY_GROUP": CREATE_ENTRY_GROUP_BOOLEAN, "CLOUD_STORAGE_BUCKET_ID": "BUCKET_ID", "SERVICE_ACCOUNT": "SERVICE_ACCOUNT_ID", "ADDITIONAL_CONNECTOR_ARGS": [ADDITIONAL_CONNECTOR_ARGUMENTS], "CUSTOM_CONTAINER_IMAGE": "CONTAINER_IMAGE", "IMPORT_JOB_SCOPE_ENTRY_TYPES": [ENTRY_TYPES], "IMPORT_JOB_SCOPE_ASPECT_TYPES": [ASPECT_TYPES], "IMPORT_JOB_LOG_LEVEL": "INFO", "NETWORK_TAGS": [], "NETWORK_URI": "", "SUBNETWORK_URI": ""}
   
   
   workflow_source                 = <<EOF
   main:
     params: [args]
     steps:
       - init:
           assign:
           - WORKFLOW_ID: $${"metadataworkflow-" + sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
           - NETWORK_URI: $${default(map.get(args, "NETWORK_URI"), "")}
           - SUBNETWORK_URI: $${default(map.get(args, "SUBNETWORK_URI"), "")}
           - NETWORK_TAGS: $${default(map.get(args, "NETWORK_TAGS"), [])}
   
       - check_networking:
           switch:
             - condition: $${NETWORK_URI != "" and SUBNETWORK_URI != ""}
               raise: "Error: cannot set both network_uri and subnetwork_uri. Please select one."
             - condition: $${NETWORK_URI != ""}
               steps:
                 - submit_extract_job_with_network_uri:
                     assign:
                       - NETWORKING: $${NETWORK_URI}
                       - NETWORK_TYPE: "networkUri"
             - condition: $${SUBNETWORK_URI != ""}
               steps:
                 - submit_extract_job_with_subnetwork_uri:
                     assign:
                       - NETWORKING: $${SUBNETWORK_URI}
                       - NETWORK_TYPE: "subnetworkUri"
           next: set_default_networking
   
       - set_default_networking:
           assign:
             - NETWORK_TYPE: "networkUri"
             - NETWORKING: $${"projects/" + args.TARGET_PROJECT_ID + "/global/networks/default"}
           next: check_create_target_entry_group
   
       - check_create_target_entry_group:
           switch:
             - condition: $${args.CREATE_TARGET_ENTRY_GROUP == true}
               next: create_target_entry_group
             - condition: $${args.CREATE_TARGET_ENTRY_GROUP == false}
               next: generate_extract_job_link
   
       - create_target_entry_group:
           call: http.post
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups?entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           next: generate_extract_job_link
   
       - generate_extract_job_link:
           call: sys.log
           args:
               data: $${"https://console.cloud.google.com/dataproc/batches/" + args.CLOUD_REGION + "/" + WORKFLOW_ID + "/monitoring?project=" + args.TARGET_PROJECT_ID}
               severity: "INFO"
           next: submit_pyspark_extract_job
   
       - submit_pyspark_extract_job:
           call: http.post
           args:
             url: $${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches"}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
             headers:
               Content-Type: "application/json"
             query:
               batchId: $${WORKFLOW_ID}
             body:
               pysparkBatch:
                 mainPythonFileUri: file:///main.py
                 args:
                   - $${"--target_project_id=" + args.TARGET_PROJECT_ID}
                   - $${"--target_location_id=" + args.CLOUD_REGION}
                   - $${"--target_entry_group_id=" + args.TARGET_ENTRY_GROUP_ID}
                   - $${"--output_bucket=" + args.CLOUD_STORAGE_BUCKET_ID}
                   - $${"--output_folder=" + WORKFLOW_ID}
                   - $${args.ADDITIONAL_CONNECTOR_ARGS}
               runtimeConfig:
                   containerImage: $${args.CUSTOM_CONTAINER_IMAGE}
               environmentConfig:
                   executionConfig:
                       serviceAccount: $${args.SERVICE_ACCOUNT}
                       stagingBucket: $${args.CLOUD_STORAGE_BUCKET_ID}
                       $${NETWORK_TYPE}: $${NETWORKING}
                       networkTags: $${NETWORK_TAGS}
           result: RESPONSE_MESSAGE
           next: check_pyspark_extract_job
   
       - check_pyspark_extract_job:
           call: http.get
           args:
             url: $${"https://dataproc.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/batches/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: PYSPARK_EXTRACT_JOB_STATUS
           next: check_pyspark_extract_job_done
   
       - check_pyspark_extract_job_done:
           switch:
             - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "SUCCEEDED"}
               next: generate_import_logs_link
             - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "CANCELLED"}
               raise: $${PYSPARK_EXTRACT_JOB_STATUS}
             - condition: $${PYSPARK_EXTRACT_JOB_STATUS.body.state == "FAILED"}
               raise: $${PYSPARK_EXTRACT_JOB_STATUS}
           next: pyspark_extract_job_wait
   
       - pyspark_extract_job_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: check_pyspark_extract_job
   
       - generate_import_logs_link:
           call: sys.log
           args:
               data: $${"https://console.cloud.google.com/logs/query?project=" + args.TARGET_PROJECT_ID + "&query=resource.type%3D%22dataplex.googleapis.com%2FMetadataJob%22+AND+resource.labels.location%3D%22" + args.CLOUD_REGION + "%22+AND+resource.labels.metadata_job_id%3D%22" + WORKFLOW_ID + "%22"}
               severity: "INFO"
           next: submit_import_job
   
       - submit_import_job:
           call: http.post
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs?metadata_job_id=" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
             body:
               type: IMPORT
               import_spec:
                 source_storage_uri: $${"gs://" + args.CLOUD_STORAGE_BUCKET_ID + "/" + WORKFLOW_ID + "/"}
                 entry_sync_mode: FULL
                 aspect_sync_mode: INCREMENTAL
                 log_level: $${default(map.get(args, "IMPORT_JOB_LOG_LEVEL"), "INFO")}
                 scope:
                   entry_groups: 
                     - $${"projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/entryGroups/" + args.TARGET_ENTRY_GROUP_ID}
                   entry_types: $${args.IMPORT_JOB_SCOPE_ENTRY_TYPES}
                   aspect_types: $${args.IMPORT_JOB_SCOPE_ASPECT_TYPES}
           result: IMPORT_JOB_RESPONSE
           next: get_job_start_time
   
       - get_job_start_time:
           assign:
             - importJobStartTime: $${sys.now()}
           next: import_job_startup_wait
   
       - import_job_startup_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: initial_get_import_job
   
       - initial_get_import_job:
           call: http.get
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: IMPORT_JOB_STATUS
           next: check_import_job_status_available
   
       - check_import_job_status_available:
           switch:
             - condition: $${sys.now() - importJobStartTime > 300}  # 5 minutes = 300 seconds
               next: kill_import_job
             - condition: $${"status" in IMPORT_JOB_STATUS.body}
               next: check_import_job_done
           next: import_job_status_wait
   
       - import_job_status_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: check_import_job_status_available
   
       - check_import_job_done:
           switch:
             - condition: $${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED"}
               next: the_end
             - condition: $${IMPORT_JOB_STATUS.body.status.state == "CANCELLED"}
               raise: $${IMPORT_JOB_STATUS}
             - condition: $${IMPORT_JOB_STATUS.body.status.state == "SUCCEEDED_WITH_ERRORS"}
               raise: $${IMPORT_JOB_STATUS}
             - condition: $${IMPORT_JOB_STATUS.body.status.state == "FAILED"}
               raise: $${IMPORT_JOB_STATUS}
             - condition: $${sys.now() - importJobStartTime > 43200}  # 12 hours = 43200 seconds
               next: kill_import_job
           next: import_job_wait
   
       - get_import_job:
           call: http.get
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: IMPORT_JOB_STATUS
           next: check_import_job_done
   
       - import_job_wait:
           call: sys.sleep
           args:
             seconds: 30
           next: get_import_job
   
       - kill_import_job:
           call: http.post
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID + ":cancel"}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           next: get_killed_import_job
   
       - get_killed_import_job:
           call: http.get
           args:
             url: $${"https://dataplex.googleapis.com/v1/projects/" + args.TARGET_PROJECT_ID + "/locations/" + args.CLOUD_REGION + "/metadataJobs/" + WORKFLOW_ID}
             auth:
               type: OAuth2
               scopes: "https://www.googleapis.com/auth/cloud-platform"
           result: KILLED_IMPORT_JOB_STATUS
           next: killed
   
       - killed:
           raise: $${KILLED_IMPORT_JOB_STATUS}
   
       - the_end:
           return: $${IMPORT_JOB_STATUS}
   EOF
   

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

   • ‫PROJECT_ID: השם של פרויקט היעד Google Cloudשאליו רוצים לייבא את המטא-נתונים.
   • ‫LOCATION_ID: מיקום היעד שבו יפעלו המשימות של Managed Service for Apache Spark ושל ייבוא המטא-נתונים, ושאליו יובאו המטא-נתונים. Google Cloud
   • ‫SERVICE_ACCOUNT_ID: חשבון השירות שהגדרתם בקטע תפקידים נדרשים במסמך הזה.
   • ‫CRON_SCHEDULE_EXPRESSION: ביטוי cron שמגדיר את לוח הזמנים להרצת צינור העיבוד. לדוגמה, כדי להפעיל את התזמון בחצות בכל יום, משתמשים בביטוי 0 0 * * *.
   • ‫ENTRY_GROUP_ID: המזהה של קבוצת הרשומות שאליה רוצים לייבא את המטא-נתונים. מזהה קבוצת הרשומות יכול להכיל אותיות קטנות, מספרים ומקפים.

     השם המלא של המשאב של קבוצת הרשומות הזו הוא 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 צריך להיות זהה למיקום שאליו מייבאים את המטא-נתונים, או global.Google Cloud
   • אופציונלי: עבור הארגומנט NETWORK_TAGS, מספקים רשימה של תגי רשת.
   • אופציונלי: בארגומנט NETWORK_URI, מציינים את ה-URI של רשת ה-VPC שמקושרת למקור הנתונים. אם מציינים רשת, לא צריך לציין את הארגומנט subnetwork.
   • אופציונלי: בארגומנט SUBNETWORK_URI מציינים את ה-URI של רשת המשנה שמקושרת למקור הנתונים. אם מציינים רשת משנה, לא צריך לציין את ארגומנט הרשת.

3. מאתחלים את Terraform:

   terraform init
   

4. מאמתים את Terraform באמצעות הקובץ .tfvars:

   terraform plan --var-file=CONNECTOR_VARIABLES_FILE.tfvars
   

   מחליפים את CONNECTOR_VARIABLES_FILE בשם של קובץ הגדרות המשתנים.

5. פורסים את Terraform באמצעות קובץ .tfvars:

   terraform apply --var-file=CONNECTOR_VARIABLES_FILE.tfvars
   

   ‫Terraform יוצר תהליך עבודה ומשימה ב-Cloud Scheduler בפרויקט שצוין. ‫Workflows מריץ את צינור עיבוד הנתונים לפי לוח הזמנים שאתם מציינים.

   בהתאם לכמות המטא-נתונים שמייבאים, יכול להיות שתהליך העבודה יימשך כמה דקות או יותר. מידע נוסף על הצגת ההתקדמות זמין במאמר גישה לתוצאות של הרצת תהליך עבודה.

   אחרי שהפייפליין סיים לפעול, אפשר לחפש את המטא-נתונים המיובאים ב-Knowledge Catalog.

צפייה ביומני המשימות

שימוש ב-Cloud Logging כדי להציג יומנים של צינור קישוריות מנוהל. מטען הייעודי (payload) של היומן כולל קישור ליומנים של משימת האצווה של Managed Service for Apache Spark ומשימת ייבוא המטא-נתונים, לפי הצורך. מידע נוסף זמין במאמר בנושא הצגת יומני זרימת עבודה.

פתרון בעיות

כדאי לנסות את ההצעות הבאות לפתרון בעיות:

• מגדירים את רמת היומן של משימת הייבוא למשימת המטא-נתונים לשימוש ברישום ביומן ברמת ניפוי הבאגים במקום ברישום ביומן ברמת המידע.
• בודקים את היומנים של משימת האצווה של Managed Service for Apache Spark (להפעלות של מחבר) ואת משימת ייבוא המטא-נתונים. מידע נוסף זמין במאמרים בנושא שליחת שאילתות ביומנים של Managed Service for Apache Spark ושליחת שאילתות ביומנים של מטא-נתונים של משימות.
• אם אי אפשר לייבא רשומה באמצעות צינור העיבוד והודעת השגיאה לא מספקת מספיק מידע, נסו ליצור רשומה בהתאמה אישית עם אותם פרטים בקבוצת רשומות לבדיקה. מידע נוסף זמין במאמר בנושא יצירת רשומה בהתאמה אישית.

המאמרים הבאים

• מידע על ניהול מטא-נתונים ב-Knowledge Catalog
• פיתוח מחבר מותאם אישית לייבוא מטא-נתונים