ייצוא נתונים ל-Pub/Sub (תהליך ETL הפוך)

חילוץ, טרנספורמציה וטעינה (ETL) הפוכים מ-BigQuery ל-Pub/Sub מאפשרים להציג נתונים לאפליקציות ולשירותים במורד הזרם באופן מבוסס-אירועים. תהליך העבודה הזה משלב את יכולות הניתוח של BigQuery עם שירות ההודעות הגלובלי האסינכרוני והסקלאבילי של Pub/Sub.

כדי לייצא נתונים ל-Pub/Sub, משתמשים בהצהרה EXPORT DATA בשאילתה מתמשכת כדי לייצא נתונים מ-BigQuery לנושא Pub/Sub.

דרישות מוקדמות

אתם צריכים ליצור חשבון שירות. כדי להריץ שאילתה מתמשכת שמייצאת תוצאות לנושא Pub/Sub, צריך חשבון שירות.

צריך ליצור נושא Pub/Sub כדי לקבל את תוצאות השאילתה המתמשכת כהודעות, ומינוי Pub/Sub שאפליקציית היעד תוכל להשתמש בו כדי לקבל את ההודעות האלה.

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

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

הרשאות בחשבון המשתמש

כדי ליצור משימה ב-BigQuery, לחשבון המשתמש צריכה להיות הרשאת ה-IAM‏ bigquery.jobs.create. כל אחד מהתפקידים הבאים ב-IAM נותן את ההרשאה bigquery.jobs.create:

כדי לשלוח משימה שפועלת באמצעות חשבון שירות, לחשבון המשתמש צריך להיות התפקיד משתמש בחשבון שירות (roles/iam.serviceAccountUser). אם אתם משתמשים באותו חשבון משתמש כדי ליצור את חשבון השירות, לחשבון המשתמש צריך להיות התפקיד אדמין של חשבון שירות (roles/iam.serviceAccountAdmin). במאמר הקצאת תפקיד יחיד מוסבר איך להגביל את הגישה של משתמש לחשבון שירות אחד, במקום לכל חשבונות השירות בפרויקט.

אם צריך להפעיל בחשבון המשתמש את ממשקי ה-API שנדרשים לתרחיש לדוגמה של השאילתה המתמשכת, צריך להקצות לחשבון המשתמש את התפקיד אדמין של שימוש בשירות (roles/serviceusage.serviceUsageAdmin).

הרשאות של חשבון שירות

כדי לייצא נתונים מטבלה ב-BigQuery, לחשבון השירות צריכה להיות הרשאת ה-IAM‏ bigquery.tables.export. כל אחד מהתפקידים הבאים ב-IAM נותן את ההרשאה bigquery.tables.export:

כדי שחשבון השירות יוכל לגשת ל-Pub/Sub, צריך להקצות לו את שני תפקידי ה-IAM הבאים:

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

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

מפעילים את BigQuery API ואת Pub/Sub API, אם הם עדיין לא מופעלים.

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

כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד "בעלים" (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד "אדמין בממשק 'שימוש בשירות'" (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים

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

ייצוא ל-Pub/Sub

כדי לייצא נתונים לנושא Pub/Sub, משתמשים בהצהרה EXPORT DATA:

המסוף

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

    כניסה ל-BigQuery

  2. בעורך השאילתות, לוחצים על Edit > Query settings.

  3. בקטע שאילתה מתמשכת, מסמנים את התיבה Use continuous query mode.

  4. בתיבה Service account, בוחרים את חשבון השירות שיצרתם.

  5. לוחצים על Save.

  6. בעורך השאילתות, מזינים את ההצהרה הבאה:

    EXPORT DATA
OPTIONS (
format = 'CLOUD_PUBSUB',
uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID'
) AS
(
QUERY
);

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

    • PROJECT_ID: מזהה הפרויקט.
    • ‫TOPIC_ID: מזהה הנושא ב-Pub/Sub. אפשר לקבל את מזהה הנושא מהדף Topics במסוף Google Cloud .
    • ‫QUERY: הצהרת ה-SQL לבחירת הנתונים לייצוא. הצהרת ה-SQL יכולה להכיל רק פעולות נתמכות. כדי לציין את נקודת הזמן שבה מתחיל עיבוד הנתונים, צריך להשתמש בפונקציה APPENDS בסעיף FROM של שאילתה מתמשכת.

  7. לוחצים על Run.

BQ

  1. מפעילים את Cloud Shell במסוף Google Cloud .

    הפעלת Cloud Shell

    בחלק התחתון של מסוף Google Cloud יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

  2. מריצים את השאילתה המתמשכת בשורת הפקודה באמצעות הפקודה bq query עם הדגלים הבאים:

    • מגדירים את הדגל --continuous לערך true כדי שהשאילתה תהיה רציפה.
    • משתמשים בדגל --connection_property כדי לציין את חשבון השירות שבו רוצים להשתמש.
    bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'EXPORT DATA OPTIONS (format = "CLOUD_PUBSUB", uri = "https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID") AS (QUERY);'

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

    • PROJECT_ID: מזהה הפרויקט.
    • ‫SERVICE_ACCOUNT_EMAIL: כתובת האימייל של חשבון השירות. אפשר למצוא את כתובת האימייל של חשבון השירות בדף Service accounts במסוף Google Cloud .
    • ‫QUERY: הצהרת ה-SQL לבחירת הנתונים לייצוא. הצהרת ה-SQL יכולה להכיל רק פעולות נתמכות. כדי לציין את נקודת הזמן שבה מתחיל עיבוד הנתונים, צריך להשתמש בפונקציה APPENDS בסעיף FROM של שאילתה מתמשכת.

API

  1. מריצים את השאילתה המתמשכת באמצעות קריאה ל-method‏ jobs.insert. מגדירים את השדות הבאים במשאב JobConfigurationQuery של משאב Job שאתם מעבירים:

    • מגדירים את השדה continuous לערך true כדי שהשאילתה תהיה רציפה.
    • בשדה connection_property מציינים את חשבון השירות שבו רוצים להשתמש.
    curl --request POST \
  'https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs'
  --header 'Authorization: Bearer $(gcloud auth print-access-token) \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/json' \
  --data '("configuration":("query":"EXPORT DATA OPTIONS (format = 'CLOUD_PUBSUB', uri = 'https://pubsub.googleapis.com/projects/PROJECT_ID/topics/TOPIC_ID') AS (QUERY);","useLegacySql":false,"continuous":true,"connectionProperties":["key": "service_account","value":"SERVICE_ACCOUNT_EMAIL"]))' \
  --compressed

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

    • PROJECT_ID: מזהה הפרויקט.
    • ‫QUERY: הצהרת ה-SQL לבחירת הנתונים לייצוא. הצהרת ה-SQL יכולה להכיל רק פעולות נתמכות. כדי לציין את נקודת הזמן שבה מתחיל עיבוד הנתונים, צריך להשתמש בפונקציה APPENDS בסעיף FROM של שאילתה מתמשכת.
    • ‫SERVICE_ACCOUNT_EMAIL: כתובת האימייל של חשבון השירות. אפשר למצוא את כתובת האימייל של חשבון השירות בדף Service accounts במסוף Google Cloud .

ייצוא של כמה עמודות ל-Pub/Sub

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

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

אופטימיזציה של הייצוא

אם נראה שביצועי המשימה של השאילתה המתמשכת מוגבלים בגלל משאבי המחשוב הזמינים, נסו להגדיל את הקצאת המערכים שהוזמנו של יחידות ה-CPU של CONTINUOUS ב-BigQuery.

מגבלות

  • הנתונים שמייצאים חייבים לכלול עמודה אחת של STRING או BYTES. אתם יכולים לבחור כל שם לעמודה.
  • כדי לייצא ל-Pub/Sub, צריך להשתמש בשאילתה מתמשכת.
  • אי אפשר להעביר סכימה לנושא Pub/Sub בשאילתה המתמשכת.
  • אי אפשר לייצא נתונים לנושא Pub/Sub שמשתמש בסכימה.
  • כשמייצאים ל-Pub/Sub, אפשר לייצא רשומות בפורמט JSON שחלק מהערכים שלהן הם NULL, אבל אי אפשר לייצא רשומות שמכילות רק ערכי NULL. כדי להחריג רשומות של NULL מתוצאות השאילתה, אפשר לכלול מסנן WHERE message IS NOT NULL בשאילתה המתמשכת.
  • כשמייצאים נתונים לנושא Pub/Sub שמוגדר עם נקודת קצה (endpoint) לפי מיקום, צריך להגדיר את נקודת הקצה באותו גבול אזורי של Google Cloud כמו מערך הנתונים ב-BigQuery שמכיל את הטבלה ששולחים אליה שאילתה.
  • הנתונים שמייצאים לא יכולים לחרוג מהמכסות של Pub/Sub.

תמחור

כשמייצאים נתונים בשאילתה מתמשכת, החיוב מתבצע לפי תמחור המחשוב של קיבולת BigQuery. כדי להריץ שאילתות מתמשכות, צריך הזמנה שמשתמשת במהדורת Enterprise או Enterprise Plus, והקצאת הזמנה שמשתמשת בסוג המשימה CONTINUOUS.

אחרי ייצוא הנתונים, תחויבו על השימוש ב-Pub/Sub. מידע נוסף זמין במאמר התמחור של Pub/Sub.