יצירת שאילתות מתמשכות

במאמר הזה מוסבר איך להריץ שאילתה מתמשכת ב-BigQuery.

שאילתות מתמשכות ב-BigQuery הן הצהרות SQL שפועלות באופן רציף ומעבדות נתונים כמעט בזמן אמת. אתם יכולים לכתוב או לייצא את שורות הפלט שנוצרו על ידי שאילתה מתמשכת ליעדים הבאים:

בחירת סוג החשבון

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

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

ההרשאות הנדרשות

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

ההרשאות שנדרשות כשמשתמשים בחשבון משתמש

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

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

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

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

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

כדי לכתוב נתונים בטבלאות מנוהלות של Apache Iceberg, בנוסף להרשאות ב-BigQuery, חשבון המשתמש צריך גם את התפקיד משתמש בחיבורים של BigQuery‏ (roles/bigquery.connectionUser) כדי להשתמש בחיבור למשאב Google Cloud שמשויך לטבלה המנוהלת של Iceberg. בנוסף, לחשבון השירות שמשויך לחיבור הזה צריכות להיות ההרשאות המתאימות (לדוגמה, roles/storage.objectUser ו-roles/storage.legacyBucketReader) בקטגוריה הבסיסית של Cloud Storage. מידע נוסף זמין במאמר יצירה ושימוש בטבלאות מנוהלות של Apache Iceberg.

ההרשאות שנדרשות כשמשתמשים בחשבון שירות

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

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

כדי ליצור משימה ב-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:

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

כדי לכתוב נתונים בטבלאות מנוהלות של Apache Iceberg, בנוסף להרשאות ב-BigQuery, לחשבון השירות צריך להיות גם התפקיד משתמש בחיבור של BigQuery‏ (roles/bigquery.connectionUser) כדי להשתמש בחיבור למשאב Google Cloud שמשויך לטבלה המנוהלת של Iceberg. בנוסף, לחשבון השירות שמשויך לחיבור הזה צריכות להיות ההרשאות המתאימות (לדוגמה, roles/storage.objectUser ו-roles/storage.legacyBucketReader) בקטגוריה הבסיסית של Cloud Storage. מידע נוסף זמין במאמר יצירה ושימוש בטבלאות מנוהלות של Apache Iceberg.

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

  1. בדף לבחירת הפרויקט במסוף Google Cloud , בוחרים פרויקט ב- Google Cloud או יוצרים אותו.

    התפקידים שנדרשים כדי לבחור או ליצור פרויקט

    • בחירת פרויקט: כדי לבחור פרויקט לא נדרש תפקיד IAM ספציפי – אתם יכולים לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד "יצירת פרויקטים" (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים?

    כניסה לדף לבחירת הפרויקט

  2. מוודאים שהחיוב מופעל בפרויקט ב- Google Cloud .

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

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

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

    להפעלת ה-API

יצירת בקשה לשמירת מקום

יוצרים הזמנה במהדורת Enterprise או Enterprise Plus, ואז יוצרים הקצאת הזמנה עם סוג משימה CONTINUOUS. בהזמנה הזו אפשר להשתמש בהתאמה לעומס (autoscaling) ובשיתוף משבצות לא פעילות. יש מגבלות שחלות על הקצאת הזמנות לשאילתות מתמשכות.

ייצוא ל-Pub/Sub

כדי לייצא נתונים ל-Pub/Sub, צריך ממשקי API, הרשאות IAM ומשאבים Google Cloud נוספים. מידע נוסף על ייצוא ל-Pub/Sub

עיבוד מוטציות באמצעות CHANGES

כשמייצאים נתונים ל-Pub/Sub, אפשר להשתמש בפונקציית היסטוריית השינויים של CHANGES. הפונקציה CHANGES מעבדת את כל השורות שהשתנו בטבלת המקור, כולל הוספות ושינויים.

הטמעת מאפיינים מותאמים אישית כמטא-נתונים בהודעות Pub/Sub

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

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

העמודה _ATTRIBUTES צריכה להיות מסוג JSON, בפורמט ARRAY<STRUCT<STRING, STRING>> או STRUCT<STRING>.

דוגמה לייצוא נתונים לנושא Pub/Sub

ייצוא ל-Bigtable

כדי לייצא נתונים ל-Bigtable, צריך ממשקי API נוספים, הרשאות IAM ומשאבים של Google Cloud. מידע נוסף זמין במאמר ייצוא ל-Bigtable.

ייצוא ל-Spanner

כדי לייצא נתונים ל-Spanner, צריך ממשקי API נוספים, הרשאות IAM ומשאבי Google Cloud. מידע נוסף זמין במאמר ייצוא ל-Spanner (‏ETL הפוך).

כתיבת נתונים בטבלה ב-BigQuery או בטבלאות מנוהלות של Apache Iceberg

אתם יכולים לכתוב נתונים בטבלה ב-BigQuery או בטבלה מנוהלת של Iceberg באמצעות הצהרת INSERT.

כשכותבים פלט של שאילתה מתמשכת בטבלה מנוהלת של Iceberg:

  • טבלת היעד המנוהלת ב-Iceberg צריכה להיות קיימת לפני שיוצרים את השאילתה המתמשכת. הוראות ליצירת טבלה מנוהלת של Iceberg מפורטות במאמר יצירה ושימוש בטבלאות מנוהלות של Apache Iceberg.
  • תחביר השאילתה זהה לכתיבה בטבלה רגילה ב-BigQuery. אפשר לציין את טבלת היעד בסעיף INSERT INTO, ואין צורך לציין אפשרויות חיבור או קטגוריה בהצהרת ה-SQL.
  • לחשבון המשתמש או לחשבון השירות שמריץ את השאילתה המתמשכת צריכות להיות ההרשאות הנדרשות בטבלת היעד, בחיבור למשאב Google Cloud ובקטגוריה הבסיסית של Cloud Storage. למידע נוסף, קראו את המאמר ההרשאות הנדרשות.

לדוגמה, ראו כתיבת נתונים בטבלה ב-BigQuery או בטבלאות מנוהלות של Apache Iceberg.

שימוש בפונקציות AI

כדי להשתמש בפונקציית AI נתמכת בשאילתה מתמשכת, נדרשים ממשקי API, הרשאות IAM ומשאבים Google Cloudנוספים. למידע נוסף, תוכלו לקרוא את אחד מהנושאים הבאים, בהתאם לתרחיש לדוגמה:

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

מציינים שעת התחלה לשאילתה המתמשכת.

כדי לציין את הנתונים המוקדמים ביותר לעיבוד, צריך להשתמש בפונקציה APPENDS, או בפונקציה CHANGES במקרה של ייצוא ל-Pub/Sub, בסעיף FROM של שאילתה מתמשכת. לדוגמה, APPENDS(TABLE my_table, start_timestamp).

הארגומנט start_timestamp מגדיר את נקודת הזמן שבה השאילתה המתמשכת מתחילה לעבד נתונים. לדוגמה, הערך APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) מורה ל-BigQuery לעבד נתונים שנוספו לטבלה my_table עד 10 דקות לפני תחילת השאילתה המתמשכת. נתונים שנוספים ל-my_table לאחר מכן מעובדים כשהם מגיעים. אין עיכוב בעיבוד הנתונים.

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

אל תספקו ארגומנט end_timestamp לפונקציה APPENDS כשאתם משתמשים בה בשאילתה מתמשכת.

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

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.taxirides`,
      -- 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');

צריך לציין נקודת התחלה שמוקדמת מחלון הזמן של הנסיעה בזמן

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

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

  1. מריצים שאילתה רגילה כדי למלא נתונים עד לנקודת זמן מסוימת:

    INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM `myproject.real_time_taxi_streaming.taxirides`
  -- Include all data inserted into the table up to this handoff point.
  -- This handoff timestamp must be within the time travel window.
  FOR SYSTEM_TIME AS OF '2025-01-01 00:00:00 UTC'
WHERE
  ride_status = 'dropoff';

  2. מריצים שאילתה מתמשכת מהנקודה בזמן שבה השאילתה נעצרה:

    INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- Configure the APPENDS TVF start_timestamp to start processing
    -- data right where the batch query left off + 1 microsecond.
    -- This timestamp must be within the time travel window.
    TIMESTAMP '2025-01-01 00:00:00 UTC' + INTERVAL 1 MICROSECOND)
WHERE
  ride_status = 'dropoff';

הרצת שאילתה מתמשכת באמצעות חשבון משתמש

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

כדי להריץ שאילתה מתמשכת:

המסוף

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

    כניסה ל-BigQuery

  2. בעורך השאילתות, לוחצים על Edit > Query mode > שאילתה מתמשכת. כדי לאשר את הבחירה, לוחצים על Confirm.

  3. אופציונלי: כדי לקבוע כמה זמן השאילתה תפעל, לוחצים על Edit > Query settings ומגדירים את Job timeout באלפיות השנייה.

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

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

BQ

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

    הפעלת Cloud Shell

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

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

    bq query --use_legacy_sql=false --continuous=true
'QUERY'

    מחליפים את QUERY בהצהרת ה-SQL של השאילתה המתמשכת. הצהרת ה-SQL יכולה להכיל רק פעולות נתמכות. אתם יכולים לקבוע כמה זמן השאילתה תפעל באמצעות הדגל --job_timeout_ms.

API

מריצים את השאילתה המתמשכת באמצעות קריאה ל-method‏ jobs.insert. צריך להגדיר את השדה continuous לערך true ב-JobConfigurationQuery של המשאב Job שמעבירים. אפשר גם לקבוע כמה זמן השאילתה תפעל באמצעות הגדרת השדה jobTimeoutMs.

curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true}}}' \
  --compressed

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

  • PROJECT_ID: מזהה הפרויקט.
  • ‫QUERY: הצהרת ה-SQL של השאילתה המתמשכת. הצהרת ה-SQL יכולה להכיל רק פעולות נתמכות.

הרצת שאילתה מתמשכת באמצעות חשבון שירות

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

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

כדי להשתמש בחשבון שירות להרצת שאילתה מתמשכת:

המסוף

  1. יצירה של חשבון שירות.
  2. נותנים לחשבון השירות את ההרשאות הנדרשות.

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

    כניסה ל-BigQuery

  4. בעורך השאילתות, לוחצים על Edit > Query mode > שאילתה מתמשכת. כדי לאשר את הבחירה, לוחצים על אישור.

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

  6. בקטע שאילתה מתמשכת, בוחרים את חשבון השירות שיצרתם בתיבה Service account.

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

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

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

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

BQ

  1. יצירה של חשבון שירות.
  2. נותנים לחשבון השירות את ההרשאות הנדרשות.

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

    הפעלת Cloud Shell

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

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

    • מגדירים את הדגל --continuous לערך true כדי שהשאילתה תהיה רציפה.
    • משתמשים בדגל --connection_property כדי לציין את חשבון השירות שבו רוצים להשתמש.
    • אופציונלי: מגדירים את הדגל --job_timeout_ms כדי להגביל את זמן הריצה של השאילתה.
    bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'QUERY'

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

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

API

  1. יצירה של חשבון שירות.
  2. נותנים לחשבון השירות את ההרשאות הנדרשות.

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

    • מגדירים את השדה continuous לערך true כדי שהשאילתה תהיה רציפה.
    • בשדה connectionProperties מציינים את חשבון השירות שבו רוצים להשתמש.

    אפשר גם לקבוע כמה זמן השאילתה תפעל, באמצעות הגדרת השדה jobTimeoutMs במשאב JobConfiguration.

    curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true,"connectionProperties":[{"key":"service_account","value":"SERVICE_ACCOUNT_EMAIL"}]}}}' \
  --compressed

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

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

יצירת מזהה משרה בהתאמה אישית

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

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

    כניסה ל-BigQuery

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

  3. לוחצים על אישור.

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

  5. בקטע Custom job ID prefix, מזינים קידומת לשם משימה ייעודית.

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

עיבוד עם שמירת מצב באמצעות JOIN וצבירת נתונים בחלונות זמן

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

למידע מפורט על השימוש בפעולות האלה, קראו את הנושאים הבאים:

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

דוגמאות

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

ייצוא נתונים לנושא Pub/Sub

בדוגמה הבאה מוצגת שאילתה מתמשכת שמסננת נתונים מטבלה ב-BigQuery, שמקבלת מידע בסטרימינג על נסיעות במונית, ומפרסמת את הנתונים של נסיעות שבוטלו בנושא Pub/Sub בזמן אמת עם מאפייני הודעות:

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,
    TO_JSON(
      STRUCT(
        CAST(passenger_comment AS STRING) AS passenger_comment))
  FROM
    CHANGES(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the CHANGES 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 _CHANGE_TYPE = 'DELETE'
);

ייצוא נתונים לטבלה ב-Bigtable

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

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_BIGTABLE',
    truncate = TRUE,
    overwrite = TRUE,
    uri = 'https://bigtable.googleapis.com/projects/myproject/instances/mybigtableinstance/tables/taxi-real-time-rides')
AS (
  SELECT
    CAST(CONCAT(ride_id, timestamp, latitude, longitude) AS STRING) AS rowkey,
    STRUCT(
      timestamp,
      latitude,
      longitude,
      meter_reading,
      ride_status,
      passenger_count) AS features
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- 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'
);

ייצוא נתונים לטבלת Spanner

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

EXPORT DATA
 OPTIONS (
   format = 'CLOUD_SPANNER',
   uri = 'https://spanner.googleapis.com/projects/myproject/instances/myspannerinstance/databases/taxi-real-time-rides',
   spanner_options ="""{
      "table": "rides",
      -- To ensure data is written to Spanner in the correct sequence
      -- during a continuous export, use the change_timestamp_column
      -- option. This should be mapped to a timestamp column from your
      -- BigQuery data. If your source data lacks a timestamp, the
      -- _CHANGE_TIMESTAMP pseudocolumn provided by the APPENDS function
      -- will be automatically mapped to the "change_timestamp" column.
      "change_timestamp_column": "change_timestamp"
   }"""
  )
  AS (
  SELECT
    ride_id,
    latitude,
    longitude,
    meter_reading,
    ride_status,
    passenger_count,
    _CHANGE_TIMESTAMP as change_timestamp
  FROM APPENDS(
        TABLE `myproject.real_time_taxi_streaming.taxirides`,
        -- 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'
  );

כתיבת נתונים בטבלה ב-BigQuery או בטבלאות מנוהלות של Apache Iceberg

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

INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- 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 = 'dropoff';

כדי לכתוב פלט של שאילתה מתמשכת בטבלה מנוהלת של Iceberg, מחליפים את myproject.real_time_taxi_streaming.transformed_taxirides בשם של טבלת היעד המנוהלת של Iceberg. התחביר זהה כי פרטי החיבור והגדרות האחסון ב-Lakehouse מוגדרים בטבלת היעד עצמה.

עיבוד נתונים באמצעות מודל של Gemini Enterprise Agent Platform

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

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,
        prompt,
        result)) AS message
  FROM
    AI.GENERATE_TEXT(
      MODEL `myproject.real_time_taxi_streaming.taxi_ml_generate_model`,
      (
        SELECT
          timestamp,
          ride_id,
          latitude,
          longitude,
          CONCAT(
            'Generate an ad based on the current latitude of ',
            latitude,
            ' and longitude of ',
            longitude) AS prompt
        FROM
          APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
            -- 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'
      ),
      STRUCT(
        50 AS max_output_tokens,
        1.0 AS temperature,
        40 AS top_k,
        1.0 AS top_p))
      AS ml_output
);

ביצוע צבירות של JOIN וחלונות

בדוגמה הבאה מוצגת שאילתה מתמשכת שמבצעת JOIN וצבירות בחלונות.

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

INSERT INTO
 `real_time_taxi_streaming.neighborhood_taxi_health`
WITH potential_matches AS (
 SELECT
   requests._CHANGE_TIMESTAMP AS bq_changed_ts,
   requests.geohash,
   requests.latitude,
   requests.longitude,
   ST_DISTANCE(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude)
   ) AS distance_in_meters
 FROM
   APPENDS(TABLE `real_time_taxi_streaming.ride_requests`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS requests
 INNER JOIN
   APPENDS(TABLE `real_time_taxi_streaming.taxirides`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS taxis
 ON requests.geohash = taxis.geohash
 WHERE
   taxis.ride_status = 'available'
   AND taxis._CHANGE_TIMESTAMP BETWEEN (requests._CHANGE_TIMESTAMP - INTERVAL 5 MINUTE) AND requests._CHANGE_TIMESTAMP
   AND ST_Dwithin(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude),
     2000 -- Distance in meters
   )
)
SELECT
 window_end,
 geohash,
 ROUND(AVG(latitude), 6) AS avg_latitude,
 ROUND(AVG(longitude), 6) AS avg_longitude,
 COUNT(*) AS taxi_demand_volume,
 ROUND(AVG(distance_in_meters), 2) AS avg_proximity_meters,
 ROUND(MIN(distance_in_meters), 2) AS min_proximity_meters,
 ROUND(MAX(distance_in_meters), 2) AS max_proximity_meters,
 ROUND(STDDEV(distance_in_meters), 2) AS proximity_stddev
FROM
 TUMBLE(TABLE potential_matches, "bq_changed_ts", INTERVAL 5 MINUTE)
GROUP BY
 window_end,
 geohash;

שינוי ה-SQL של שאילתה מתמשכת

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

כדי לשנות את ה-SQL שמשמש בשאילתה מתמשכת:

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

  4. משיגים את הערך של end_time למשימת השאילתה המתמשכת המקורית באמצעות INFORMATION_SCHEMA JOBS view:

    SELECT end_time
FROM `PROJECT_ID.region-REGION`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE
  EXTRACT(DATE FROM creation_time) = current_date()
AND error_result.reason = 'stopped'
AND job_id = 'JOB_ID';

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

    • PROJECT_ID: מזהה הפרויקט.
    • ‫REGION: האזור שבו משתמשים בפרויקט.
    • ‫JOB_ID: מזהה המשימה של השאילתה המתמשכת שזיהיתם בשלב 1.

  5. משנים את הצהרת ה-SQL של השאילתה המתמשכת כדי להתחיל את השאילתה המתמשכת מנקודת זמן מסוימת, ומשתמשים בערך end_time שקיבלתם בשלב 5 כנקודת ההתחלה.

  6. משנים את הצהרת ה-SQL של השאילתה המתמשכת בהתאם לשינויים שרוצים לבצע.

  7. מריצים את השאילתה המתמשכת ששיניתם.

ביטול שאילתה מתמשכת

אפשר לבטל משימת שאילתה מתמשכת כמו כל משימה אחרת. יכול להיות שיעבור עד דקה עד שהשאילתה תפסיק לפעול אחרי שהמשימה תבוטל.

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

מעקב אחרי שאילתות וטיפול בשגיאות

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

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