מעקב אחרי סטטוס הפייפליין

בדף הזה מוסבר איך לפרסם אירועים של צינורות Cloud Data Fusion, כמו סטטוס של צינור, בנושאי Pub/Sub. בנוסף, מוסבר איך ליצור פונקציות Cloud Run שמעבדות את ההודעות ב-Pub/Sub ומבצעות פעולות, כמו זיהוי של צינורות שנכשלו וניסיון חוזר להפעיל אותם.

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

• יוצרים נושא שבו Pub/Sub יכול לפרסם אירועים של פייפליין Cloud Data Fusion.

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

כדי לוודא שלחשבון השירות של Cloud Data Fusion יש את ההרשאות הנדרשות לפרסום אירועים של צינורות בנושא Pub/Sub, צריך לבקש מהאדמין להקצות לחשבון השירות של Cloud Data Fusion את תפקיד ה-IAM ‏Pub/Sub Publisher (roles/pubsub.publisher) בפרויקט שבו יוצרים את נושא ה-Pub/Sub.

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

מידע נוסף על הקצאת תפקידים מופיע במאמר ניהול הגישה לפרויקטים, לתיקיות ולארגונים.

יכול להיות שהאדמין גם יוכל לתת לחשבון השירות של Cloud Data Fusion את ההרשאות שנדרשות באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

ניהול פרסום אירועים במופע של Cloud Data Fusion

אפשר לנהל את פרסום האירועים במופעים חדשים וקיימים של Cloud Data Fusion באמצעות API בארכיטקטורת REST בגרסאות 6.7.0 ואילך.

פרסום אירועים במופע חדש

יוצרים מופע חדש וכוללים בו את השדה EventPublishConfig. מידע נוסף על השדות הנדרשים למופעים חדשים זמין במאמר בנושא משאבי מופעים.

curl -X POST \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  "https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances?instanceId=INSTANCE_ID" \
  -d '{
    "version": "VERSION_NUMBER",
    "event_publish_config": {
      "enabled": true,
      "topic": "projects/PROJECT_ID/topics/TOPIC_ID"
    }
  }'

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

• ‫PROJECT_ID: מזהה הפרויקט Google Cloud
• ‫LOCATION: המיקום של הפרויקט
• ‫INSTANCE_ID: המזהה של מופע Cloud Data Fusion
• ‫VERSION_NUMBER: הגרסה של Cloud Data Fusion שבה יצרתם את המופע – לדוגמה, 6.10.1
• ‫TOPIC_ID: המזהה של נושא ה-Pub/Sub

הפעלת פרסום אירועים במכונת Cloud Data Fusion קיימת

מעדכנים את השדה EventPublishConfig במכונת Cloud Data Fusion קיימת:

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances/INSTANCE_ID?updateMask=event_publish_config \
  -d '{
    "event_publish_config": {
      "enabled": true,
      "topic": "projects/PROJECT_ID/topics/TOPIC_ID"
    }
}'

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

• ‫PROJECT_ID: מזהה הפרויקט Google Cloud
• ‫LOCATION: המיקום של הפרויקט
• ‫INSTANCE_ID: המזהה של מופע Cloud Data Fusion
• ‫TOPIC_ID: המזהה של נושא ה-Pub/Sub

הסרת פרסום אירועים ממופע

כדי להסיר פרסום אירועים ממופע, מעדכנים את הערך של פרסום האירועים enabled ל-false:

curl -X PATCH \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \ "https://datafusion.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/instances/INSTANCE_ID?updateMask=event_publish_config" \
  -d '{ "event_publish_config": { "enabled": false } }'

יצירת פונקציות לקריאת הודעות Pub/Sub

פונקציות Cloud Run יכולות לקרוא הודעות Pub/Sub ולבצע פעולות בהתאם להן, כמו ניסיון חוזר של צינורות שלא הצליחו. כדי ליצור פונקציות Cloud Run:

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

   כניסה לדף Cloud Run functions

2. לוחצים על יצירת פונקציה.

3. מזינים שם פונקציה ואזור.

4. בשדה Trigger type, בוחרים באפשרות Cloud Pub/Sub.

5. מזינים את מזהה הנושא ב-Pub/Sub.

6. לוחצים על הבא.

7. מוסיפים פונקציות לקריאת ההודעות ב-Pub/Sub ולביצוע פעולות אחרות. לדוגמה, אתם יכולים להוסיף פונקציות לתרחישי השימוש הבאים:

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

   דוגמאות לפונקציות Cloud Run מופיעות בקטע תרחיש לדוגמה.

8. לוחצים על פריסה. מידע נוסף זמין במאמר פריסת פונקציה של Cloud Run.

תרחיש לדוגמה: סטטוס של צינור עיבוד מסמכים וניסיון חוזר להפעלת צינורות עיבוד שנכשלו

בדוגמה הבאה, פונקציות Cloud Run קוראות הודעות Pub/Sub לגבי סטטוס ההרצה של הפייפליין, ואז מנסות שוב להריץ את הפייפליינים שנכשלו ב-Cloud Data Fusion.

הפונקציות לדוגמה האלה מתייחסות לרכיבים הבאים של Google Cloud :

• ‫Google Cloud project: הפרויקט שבו נוצרות פונקציות Cloud Run ונושאי Pub/Sub
• נושא Pub/Sub: נושא Pub/Sub שמקושר למכונת Cloud Data Fusion
• מופע של Cloud Data Fusion: מופע Cloud Data Fusion שבו מעצבים ומריצים צינורות
• טבלה ב-BigQuery: הטבלה ב-BigQuery שמתעדת את סטטוס הפייפליין ואת הפרטים של ההרצה וההרצה מחדש
• פונקציית Cloud Run: פונקציית Cloud Run שבה פורסים את הקוד שמנסה להפעיל מחדש צינורות שלא הצליחו

1. בדוגמה הבאה של פונקציית Cloud Run, מתבצעת קריאה של הודעות Pub/Sub לגבי אירועים של סטטוס Cloud Data Fusion.

   # Triggered from a message on a Pub/Sub topic.
   @functions_framework.cloud_event
   def cdf_event_trigger(cloud_event):
   
   decoded_message = base64.b64decode(cloud_event.data["message"]["data"]).decode('utf-8') # Decode Pub/Sub message.
   pubsub_message = json.loads(decoded_message)
   
   # Extract pipeline run details.
   projectName = pubsub_message["projectName"]
   publishTime = pubsub_message["publishTime"]
   instanceName = pubsub_message["instanceName"]
   namespace = pubsub_message["programStatusEventDetails"]["namespace"]
   applicationName = pubsub_message["programStatusEventDetails"]["applicationName"]
   status = pubsub_message["programStatusEventDetails"]["status"]
   event_timestamp = pd.to_datetime(pubsub_message["programStatusEventDetails"]["eventTime"], unit = 'ms')
   
   print(f"projectName: {projectName}")
   print(f"publishTime: {publishTime}")
   print(f"instanceName: {instanceName}")
   print(f"namespace: {namespace}")
   print(f"applicationName: {applicationName}")
   print(f"status: {status}")
   print(f"event timestamp: {event_timestamp}")
   try:
       error = pubsub_message["programStatusEventDetails"]["error"]
       print(f"error: {error}")
   except:
       print(f"Pipeline: {applicationName}'s current status: {status}")
   

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

   # Global variables.
   pipeline_rerun_count = 0
   has_pipeline_failed_and_rerun_recently = False # Timeframe: within last 60 minutes.
   table_id = "bigquery-table-1" # The BigQuery target table for storing pipeline run information.
   
   # Update BigQuery table with the pipeline status and rerun details.
   schema=[
       bigquery.SchemaField("Project_Name", "STRING"),
       bigquery.SchemaField("Instance_Name", "STRING"),
       bigquery.SchemaField("Namespace", "STRING"),
       bigquery.SchemaField("Pipeline_Name", "STRING"),
       bigquery.SchemaField("Pipeline_Status", "STRING"),
       bigquery.SchemaField("Event_Timestamp", "TIMESTAMP"),
       bigquery.SchemaField("Pipeline_Rerun_Count", "INTEGER"),
   ]
   
   # Prepare DataFrame to load the data in BigQuery.
   data = {'Project_Name':[projectName], 'Instance_Name':[instanceName], 'Namespace':[namespace], 'Pipeline_Name':[applicationName], 'Pipeline_Status':[status], 'Event_Timestamp':[event_timestamp], 'Pipeline_Rerun_Count':[pipeline_rerun_count]}
   dataframe = pd.DataFrame(data)
   
   # Prepare BigQuery data load job configuration.
   job_config = bigquery.LoadJobConfig(schema=schema)
   
   job = bq_client.load_table_from_dataframe(dataframe, table_id, job_config=job_config)
   job.result()  # Wait for the job to complete.
   
   table = bq_client.get_table(table_id)  # Make an API request.
   print("BigQuery table: {} updated.".format(table_id))
   

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

   bq_client = bigquery.Client()
   
   if status == "FAILED":
       print(f"ALERT -- Pipeline: {applicationName} has failed. Checking for rerun: pipeline hasn't failed and rerun in the last 60 minutes.")
   
       QUERY = f"""
           SELECT * FROM `{table_id}`
           WHERE Pipeline_Name = "{applicationName}" AND Pipeline_Status = "FAILED"
           AND "{event_timestamp}" < DATETIME_ADD(Event_Timestamp, INTERVAL 60 MINUTE)
           AND Pipeline_Rerun_Count > 0
           """
   
       query_job = bq_client.query_and_wait(QUERY)  # API request.
       row_count = query_job.total_rows  # Waits for query to finish.
       print(f"Query job result row count: {row_count}")
   
       if (row_count > 0):
           print("Pipeline has FAILED and rerun recently...")
           global has_pipeline_failed_and_rerun_recently
           has_pipeline_failed_and_rerun_recently = True
   

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

   if not has_pipeline_failed_and_rerun_recently:
       applicationName = applicationName
       auth_token = get_access_token()
       post_headers = {"Authorization": "Bearer " + auth_token,"Accept": "application/json"}
       cdap_endpoint = "https://instance1-project1-dot-location1.datafusion.googleusercontent.com/api"
       run_pipeline_endpoint = cdap_endpoint + "/v3/namespaces/{}/apps/{}/workflows/DataPipelineWorkflow/start".format(namespace, applicationName)
   
       # Start the job.
       response = requests.post(run_pipeline_endpoint,headers=post_headers)
       print(f"Response for restarting the failed pipeline: {response}")
       global pipeline_rerun_count
       pipeline_rerun_count = 1
   

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

• איך כותבים פונקציות Cloud Run