החל מ-10 באפריל 2026, Dataplex Universal Catalog נקרא עכשיו Knowledge Catalog. השמות של ה-API, ספריית הלקוח, ה-CLI ו-IAM לא משתנים. מידע נוסף זמין במאמר הכרות עם קטלוג הידע של Google Cloud.

‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

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

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

סקירה כללית על קישוריות מנוהלת

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

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

יצירת מחבר

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

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

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

איך מגדירים משאבים ב- Google Cloud

מפעילים את ממשקי ה-API הבאים, אם הם עדיין לא מופעלים: Workflows,‏ Dataproc,‏ Cloud Storage,‏ Dataplex,‏ Secret Manager,‏ Artifact Registry ו-Cloud Scheduler.
התפקידים שנדרשים כדי להפעיל ממשקי API
כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד "בעלים" (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד "אדמין בממשק 'שימוש בשירות'" (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים
הפעלת ממשקי ה-API

אם אתם לא מתכננים להריץ את צינור עיבוד הנתונים לפי לוח זמנים, אתם לא צריכים להפעיל את Cloud Scheduler API.
יוצרים סודות ב-Secret Manager כדי לאחסן את פרטי הכניסה למקור הנתונים של הצד השלישי.
מגדירים את רשת ה-VPC כדי להריץ עומסי עבודה של Managed Service for Apache Spark.
יוצרים קטגוריה של Cloud Storage כדי לאחסן את קובצי הייבוא של המטא-נתונים.
יוצרים את המשאבים הבאים של Knowledge Catalog:
1. יוצרים סוגי היבטים בהתאמה אישית לרשומות שרוצים לייבא.
2. יוצרים סוגי רשומות בהתאמה אישית לרשומות שרוצים לייבא.

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

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

כדי להריץ את צינור עיבוד הנתונים המנוהל של הקישוריות, אתם יכולים להשתמש בחשבון השירות שמוגדר כברירת מחדל ב-Compute Engine‏ (PROJECT_NUMBER-compute@developer.gserviceaccount.com) או ליצור חשבון שירות משלכם (או כמה חשבונות).

המסוף

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

כניסה לדף IAM
בוחרים את הפרויקט שאליו רוצים לייבא מטא-נתונים.
לוחצים על ‎ Grant Access‎ וממלאים את כתובת האימייל של חשבון השירות.
מקצים לחשבון השירות את התפקידים הבאים:
- כתיבת יומנים (roles/logging.logWriter)
- בעלים של קבוצת רשומות ב-Dataplex ‏ (roles/dataplex.entryGroupOwner)
- בעלים של משימת מטא-נתונים ב-Dataplex ‏ (roles/dataplex.metadataJobOwner)
- עריכה ב-Dataplex Catalog ‏ (roles/dataplex.catalogEditor)
- עריכה ב-Dataproc‏ (roles/dataproc.editor)
- Dataproc Worker ‏ (roles/dataproc.worker)
- Secret Manager Secret Accessor ‏ (roles/secretmanager.secretAccessor) – בסוד שבו מאוחסנים פרטי הכניסה למקור הנתונים
- משתמש באובייקטים באחסון (roles/storage.objectUser) – בקטגוריה של Cloud Storage
- קורא Artifact Registry ‏ (roles/artifactregistry.reader) – במאגר Artifact Registry שמכיל את קובץ האימג' של המחבר
- משתמש בחשבון שירות (roles/iam.serviceAccountUser) – אם אתם משתמשים בחשבונות שירות שונים, הקצו את התפקיד הזה לחשבון השירות שמריץ את Workflows בחשבון השירות שמריץ את המשימות באצווה של Managed Service for Apache Spark.
- מפעיל של Workflows ‏ (roles/workflows.invoker) – אם רוצים לתזמן את צינור עיבוד הנתונים
שומרים את השינויים.

gcloud

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

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="serviceAccount: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="serviceAccount:SERVICE_ACCOUNT_ID" \
    --role='roles/iam.serviceAccountUser'
```
אם אתם משתמשים בחשבונות שירות שונים, הערך של הדגל --member הוא חשבון השירות שמריץ את המשימות באצווה של Managed Service for Apache Spark.
אם רוצים לתזמן את צינור עיבוד הנתונים, צריך להקצות לחשבון השירות את התפקיד Workflows Invoker ‏ (roles/workflows.invoker):
```
gcloud projects add-iam-policy-binding PROJECT_ID \
 --member="serviceAccount:SERVICE_ACCOUNT_ID" \
 --role=roles/workflows.invoker
```

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

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

המסוף

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

חשבון שירות: חשבון השירות שהגדרתם בקטע התפקידים הנדרשים במאמר הזה.
הצפנה: בוחרים באפשרות 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}

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

מציינים את הארגומנטים הבאים של זמן הריצה:
```
{
    "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: מיקום היעד Google Cloud שבו יפעלו המשימות של Managed Service for Apache Spark וייבוא המטא-נתונים, ואליו ייובאו המטא-נתונים.
- ‫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 צריך להיות זהה למיקום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_URI, מציינים את ה-URI של רשת המשנה שמתחברת למקור הנתונים. אם מציינים רשת משנה, צריך להשמיט את הארגומנט של הרשת.
בהתאם לכמות המטא-נתונים שאתם מייבאים, יכול להיות שצינור עיבוד הנתונים יפעל במשך כמה דקות או יותר. למידע נוסף על הצגת ההתקדמות, ראו גישה לתוצאות של הפעלת תהליך עבודה.

אחרי שהצינור סיים לפעול, אפשר לחפש את המטא-נתונים שיובאו ב-Knowledge Catalog.
אופציונלי: אם רוצים להריץ את צינור עיבוד הנתונים לפי לוח זמנים, יוצרים לוח זמנים באמצעות Cloud Scheduler. ממלאים את הפרטים הבאים:
- ‫Frequency: ביטוי Unix-cron שמגדיר את לוח הזמנים להרצת צינור עיבוד הנתונים.
- ‫Workflow argument: הארגומנטים של זמן הריצה של המחבר, כמו שמתואר בשלב הקודם.
- חשבון שירות: חשבון השירות. חשבון השירות מנהל את המתזמן.

gcloud

שומרים את הגדרת עומס העבודה הבאה כקובץ 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}

מגדירים משתני 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_URI, מציינים את ה-URI של רשת המשנה שמתחברת למקור הנתונים. אם מציינים רשת משנה, צריך להשמיט את הארגומנט של הרשת.
- ‫CRON_SCHEDULE_EXPRESSION: ביטוי cron שמגדיר את לוח הזמנים להרצת צינור עיבוד הנתונים. לדוגמה, כדי להריץ את התזמון בחצות בכל יום, משתמשים בביטוי 0 0 * * *.
כדי להריץ את צינור עיבוד הנתונים על פי דרישה, מפעילים את תהליך העבודה:
```
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

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

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

עורכים את הקובץ .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: מיקום היעד Google Cloud שבו יפעלו המשימות של Managed Service for Apache Spark וייבוא המטא-נתונים, ואליו ייובאו המטא-נתונים.
‫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 צריך להיות זהה למיקוםGoogle Cloud שאליו מייבאים את המטא-נתונים, או global.
אופציונלי: בארגומנט NETWORK_TAGS, מספקים רשימה של תגי רשת.
אופציונלי: בארגומנט NETWORK_URI, מציינים את ה-URI של רשת ה-VPC שמחוברת למקור הנתונים. אם מציינים רשת, משמיטים את הארגומנט של רשת המשנה.
אופציונלי: בארגומנט SUBNETWORK_URI, מציינים את ה-URI של רשת המשנה שמתחברת למקור הנתונים. אם מציינים רשת משנה, צריך להשמיט את הארגומנט של הרשת.

מאתחלים את Terraform:
```
terraform init
```
מאמתים את Terraform באמצעות הקובץ .tfvars:
```
terraform plan --var-file=CONNECTOR_VARIABLES_FILE.tfvars
```
מחליפים את CONNECTOR_VARIABLES_FILE בשם של קובץ הגדרות המשתנים.
פורסים את 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 ושליחת שאילתות ביומני משימות של מטא-נתונים.
אם אי אפשר לייבא רשומה באמצעות צינור עיבוד הנתונים והודעת השגיאה לא מספקת מספיק מידע, נסו ליצור רשומה בהתאמה אישית עם אותם פרטים בקבוצת רשומות לבדיקה. מידע נוסף זמין במאמר יצירת רשומה בהתאמה אישית.

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

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

יצירת מחבר

איך מגדירים משאבים ב- Google Cloud

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

המסוף

gcloud

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

המסוף

gcloud

Terraform

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

פתרון בעיות

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

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