יצירת נושא ייבוא ב-Cloud Storage

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

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

מידע נוסף על Cloud Storage זמין במאמרי העזרה של Cloud Storage.

מידע נוסף על ייבוא נושאים זמין במאמר מידע על ייבוא נושאים.

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

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

כדי לקבל את ההרשאות שנדרשות ליצירה ולניהול של נושא ייבוא מ-Cloud Storage, בקשו מהאדמין להקצות לכם את התפקיד עריכת Pub/Sub (roles/pubsub.editor) ב-IAM בנושא או בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

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

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

כדי ליצור ולנהל נושא ייבוא של Cloud Storage, צריך את ההרשאות הבאות:

  • יוצרים נושא לייבוא: pubsub.topics.create
  • מחיקת נושא ייבוא: pubsub.topics.delete
  • קבלת נושא ייבוא: pubsub.topics.get
  • הצגת נושא ייבוא: pubsub.topics.list
  • פרסום בנושא ייבוא: pubsub.topics.publish
  • כדי לעדכן נושא ייבוא: pubsub.topics.update
  • אחזור מדיניות IAM לנושא ייבוא: pubsub.topics.getIamPolicy
  • מגדירים את מדיניות ה-IAM בנושא ייבוא: pubsub.topics.setIamPolicy

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

אפשר להגדיר בקרת גישה ברמת הפרויקט וברמת המשאב הבודד.

המדיניות בנושא אחסון הודעות תואמת למיקום הקטגוריה

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

  • לקטגוריות עם סוג המיקום 'אזור': המדיניות חייבת לכלול את האזור הספציפי הזה. לדוגמה, אם הקטגוריה נמצאת באזור us-central1, מדיניות אחסון ההודעות צריכה לכלול גם את us-central1.

  • לקטגוריות עם סוג מיקום של שני אזורים או מספר אזורים: המדיניות חייבת לכלול לפחות אזור אחד במיקום של שני אזורים או מספר אזורים. לדוגמה, אם הקטגוריה נמצאת ב-US multi-region, מדיניות אחסון ההודעות יכולה לכלול את us-central1, את us-east1 או כל אזור אחר ב-US multi-region.

    אם המדיניות לא כוללת את האזור של הקטגוריה, יצירת הנושא תיכשל. לדוגמה, אם הקטגוריה נמצאת ב-europe-west1 ומדיניות אחסון ההודעות כוללת רק את asia-east1, תתקבל הודעת שגיאה.

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

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

הפעלת הפרסום

כדי לאפשר פרסום, צריך להקצות את התפקיד "פרסום הודעות ב-Pub/Sub" לחשבון השירות של Pub/Sub, כדי ש-Pub/Sub יוכל לפרסם בנושא הייבוא של Cloud Storage.

הפעלת פרסום בכל נושאי הייבוא של Cloud Storage

בוחרים באפשרות הזו אם אין נושא ייבוא של Cloud Storage בפרויקט.

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

    כניסה לדף IAM

  2. מסמנים את התיבה Include Google-provided role grants.

  3. מחפשים את חשבון השירות של Pub/Sub בפורמט הבא:

    service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com

  4. לוחצים על הלחצן Edit Principal (עריכת חשבון המשתמש) בחשבון השירות הזה.

  5. אם צריך, לוחצים על Add another role.

  6. מחפשים ובוחרים את התפקיד 'פרסום הודעות ב-Pub/Sub' (roles/pubsub.publisher).

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

הפעלת פרסום בנושא יבוא יחיד ב-Cloud Storage

כדי להעניק ל-Pub/Sub הרשאה לפרסם בנושא ספציפי של ייבוא ל-Cloud Storage שכבר קיים, פועלים לפי השלבים הבאים:

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

    הפעלת Cloud Shell

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

  2. מריצים את הפקודה gcloud pubsub topics add-iam-policy-binding:

    gcloud pubsub topics add-iam-policy-binding TOPIC_ID\
   --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com"\
   --role="roles/pubsub.publisher"

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

    • ‫TOPIC_ID הוא המזהה או השם של נושא הייבוא של Cloud Storage.

    • ‫PROJECT_NUMBER הוא מספר הפרויקט. במאמר זיהוי פרויקטים מוסבר איך לראות את מספר הפרויקט.

הקצאת תפקידים ב-Cloud Storage לחשבון השירות של Pub/Sub

כדי ליצור נושא לייבוא מ-Cloud Storage, לחשבון השירות של Pub/Sub צריכה להיות הרשאת קריאה מקטגוריית Cloud Storage ספציפית. נדרשות ההרשאות הבאות:

  • storage.objects.list
  • storage.objects.get
  • storage.buckets.get

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

  • הענקת הרשאות ברמת הקטגוריה. בקטגוריה הספציפית של Cloud Storage, מקצים לחשבון השירות של Pub/Sub את התפקידים "קריאת אובייקטים באחסון מדור קודם" (roles/storage.legacyObjectReader) ו"קריאה בקטגוריה באחסון מדור קודם" (roles/storage.legacyBucketReader).

  • אם אתם חייבים להקצות תפקידים ברמת הפרויקט, אתם יכולים במקום זאת להקצות את התפקיד 'אדמין לניהול אחסון' (roles/storage.admin) בפרויקט שמכיל את הקטגוריה של Cloud Storage. מקצים את התפקיד הזה לחשבון השירות של Pub/Sub.

הרשאות של קטגוריות

כדי להקצות את התפקידים 'קריאת אובייקטים באחסון מדור קודם' (roles/storage.legacyObjectReader) ו'קריאה בקטגוריה באחסון מדור קודם' (roles/storage.legacyBucketReader) לחשבון השירות של Pub/Sub ברמת הקטגוריה, מבצעים את השלבים הבאים:

  1. פותחים את הדף Cloud Storage במסוף Google Cloud .

    כניסה ל-Cloud Storage

  2. לוחצים על הקטגוריה של Cloud Storage שממנה רוצים לקרוא הודעות ולייבא אותן לנושא הייבוא של Cloud Storage.

    הדף Bucket details נפתח.

  3. בדף Bucket details, לוחצים על הכרטיסייה Permissions.

  4. בכרטיסייה Permissions > View by Principals, לוחצים על Grant access.

    נפתח הדף הענקת גישה.

  5. בקטע Add Principals, מזינים את השם של חשבון השירות של Pub/Sub.

    הפורמט של חשבון השירות הוא service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com. לדוגמה, אם PROJECT_NUMBER=112233445566, חשבון השירות יהיה בפורמט service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com.

  6. בתפריט הנפתח Assign roles > Select a role, מזינים Object Reader ובוחרים את התפקיד Storage Legacy Object Reader.

  7. לוחצים על הוספת תפקיד נוסף.

  8. בתפריט הנפתח Select a role, מזינים Bucket Reader ובוחרים את התפקיד Storage Legacy Bucket Reader.

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

הרשאות לפרויקט

כדי להקצות את התפקיד Storage Admin ‏(roles/storage.admin) ברמת הפרויקט, מבצעים את השלבים הבאים:

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

    כניסה לדף IAM

  2. בכרטיסייה Permissions > View by Principals, לוחצים על Grant access.

    נפתח הדף הענקת גישה.

  3. בקטע Add Principals (הוספת חשבונות משתמשים), מזינים את השם של חשבון השירות של Pub/Sub.

    הפורמט של חשבון השירות הוא service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com. לדוגמה, אם PROJECT_NUMBER=112233445566, חשבון השירות יהיה בפורמט service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. בתפריט הנפתח Assign roles > Select a role, מזינים Storage Admin ובוחרים את התפקיד Storage Admin.

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

למידע נוסף על IAM ב-Cloud Storage, קראו את המאמר ניהול זהויות והרשאות גישה ב-Cloud Storage.

מאפיינים של נושאים לייבוא ב-Cloud Storage

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

שם הקטגוריה

זהו השם של קטגוריה של Cloud Storage שממנה Pub/Sub קורא את הנתונים שמתפרסמים בנושא ייבוא של Cloud Storage.

פורמט קלט

כשיוצרים נושא ייבוא של Cloud Storage, אפשר לציין את הפורמט של האובייקטים שייקלטו כ-Text,‏ Avro או Pub/Sub Avro.

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

    תו מפריד. אפשר גם לציין תו מפריד שלפיו האובייקטים יפוצלו להודעות. אם לא מגדירים תו מפריד, ברירת המחדל היא תו מעבר לשורה (\n). התו המפריד חייב להיות תו יחיד.

  • Avro. האובייקטים הם בפורמט Apache Avro binary. כל אובייקט שלא בפורמט תקין של Apache Avro לא נטמע. אלה המגבלות שקשורות ל-Avro:

    • אין תמיכה בגרסאות 1.1.0 ו-1.2.0 של Avro.
    • הגודל המקסימלי של בלוק Avro הוא 16MB.

  • ‫Pub/Sub Avro. האובייקטים הם בפורמט הבינארי של Apache Avro עם סכימה שתואמת לסכימה של אובייקט שנכתב ב-Cloud Storage באמצעות מינוי ל-Cloud Storage ב-Pub/Sub עם פורמט הקובץ Avro. ריכזנו כאן כמה הנחיות חשובות לגבי Pub/Sub Avro:

    • שדה הנתונים של רשומת Avro משמש לאכלוס שדה הנתונים של הודעת Pub/Sub שנוצרה.

    • אם האפשרות write_metadata צוינה במינוי ל-Cloud Storage, כל הערכים בשדה המאפיינים יאוכלסו כמאפיינים של הודעת Pub/Sub שנוצרה.

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

זמן יצירת האובייקט המינימלי

כשיוצרים נושא לייבוא ב-Cloud Storage, אפשר לציין זמן מינימלי ליצירת אובייקט. רק אובייקטים שנוצרו בתאריך ובשעה שמופיעים בחותמת הזמן הזו או אחריהם ייכללו בייבוא. חותמת הזמן הזו צריכה להיות בפורמט כמו YYYY-MM-DDThh:mm:ssZ. כל תאריך, בעבר או בעתיד, בין 0001-01-01T00:00:00Z ל-9999-12-31T23:59:59Z כולל, הוא תקין.

התאמה לתבנית glob

כשיוצרים נושא ייבוא של Cloud Storage, אפשר לציין אופציונלית תבנית glob להתאמה. רק אובייקטים עם שמות שתואמים לתבנית הזו נטמעים. לדוגמה, כדי להטמיע את כל האובייקטים עם הסיומת .txt, אפשר לציין את תבנית ה-glob‏ **.txt.

מידע נוסף על התחביר הנתמך של תבניות glob מופיע במסמכי התיעוד של Cloud Storage.

שימוש בנושאי ייבוא של Cloud Storage

אתם יכולים ליצור נושא ייבוא חדש או לערוך נושא קיים.

לתשומת ליבכם

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

יצירת נושא ייבוא ב-Cloud Storage

כדי ליצור נושא לייבוא מ-Cloud Storage:

המסוף

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

    לדף Topics

  2. לוחצים על Create topic.

    דף הפרטים של הנושא נפתח.

  3. בשדה Topic ID (מזהה הנושא), מזינים מזהה לנושא הייבוא של Cloud Storage.

    למידע נוסף על מתן שמות לנושאים, ראו הנחיות למתן שמות.

  4. בוחרים באפשרות הוספת מינוי ברירת מחדל.

  5. לוחצים על הפעלת הטמעה.

  6. בקטע 'מקור ההטמעה', בוחרים באפשרות Google Cloud Storage.

  7. בקטע קטגוריה של Cloud Storage, לוחצים על Browse.

    הדף Select bucket נפתח. בוחרים באחת מהאפשרויות הבאות:

    • בוחרים דלי קיים מכל פרויקט מתאים.

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

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

    Unable to verify if the Pub/Sub service agent has write permissions on this bucket. You may be lacking permissions to view or set permissions.

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

  9. בשדה Object format, בוחרים באפשרות Text,‏ Avro או Pub/Sub Avro.

    אם בוחרים באפשרות Text, אפשר לציין Delimiter כדי לפצל אובייקטים להודעות.

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

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

    למידע נוסף, ראו זמן מינימלי ליצירת אובייקט.

  11. צריך לציין תבנית Glob. כדי להטמיע את כל האובייקטים בקטגוריה, משתמשים ב-** בתור תבנית ה-glob. אם המאפיין מוגדר, המערכת מבצעת המרה רק של אובייקטים שתואמים לתבנית שצוינה.

    מידע נוסף מופיע במאמר בנושא התאמה של תבנית glob.

  12. משאירים את שאר הגדרות ברירת המחדל.

  13. לוחצים על יצירת נושא.

gcloud

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

    הפעלת Cloud Shell

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

  2. מריצים את הפקודה gcloud pubsub topics create:

    gcloud pubsub topics create TOPIC_ID \
    --cloud-storage-ingestion-bucket=BUCKET_NAME \
    --cloud-storage-ingestion-input-format=INPUT_FORMAT \
    --cloud-storage-ingestion-text-delimiter=TEXT_DELIMITER \
    --cloud-storage-ingestion-minimum-object-create-time=MINIMUM_OBJECT_CREATE_TIME \
    --cloud-storage-ingestion-match-glob=MATCH_GLOB

    בפקודה, רק TOPIC_ID, הדגל --cloud-storage-ingestion-bucket והדגל --cloud-storage-ingestion-input-format נדרשים. שאר הדגלים הם אופציונליים ואפשר להשמיט אותם.

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

    • ‫TOPIC_ID: השם או המזהה של הנושא.
    • ‫BUCKET_NAME: ציון השם של קטגוריה קיימת. לדוגמה, prod_bucket. שם הקטגוריה לא יכול לכלול את מזהה הפרויקט. במאמר יצירת קטגוריות מוסבר איך ליצור קטגוריה.
    • ‫INPUT_FORMAT: מציין את הפורמט של האובייקטים שמועברים. הערך יכול להיות text, avro או pubsub_avro. מידע נוסף על האפשרויות האלה זמין במאמר בנושא פורמט קלט.

    • ‫TEXT_DELIMITER: מציין את התו שמגדיר את סוף השורה, שבאמצעותו יפוצלו אובייקטים של טקסט להודעות Pub/Sub. התו המפריד צריך להיות תו יחיד, וצריך להגדיר אותו רק אם הערך של INPUT_FORMAT הוא text. ברירת המחדל היא תו השורה החדשה (\n).

      כשמשתמשים ב-CLI של gcloud כדי לציין את התו שמפריד בין הערכים, צריך לשים לב במיוחד לטיפול בתווים מיוחדים כמו מעבר שורה \n. כדי לוודא שהתו שמפריד בין הערכים יפורש בצורה נכונה, צריך להשתמש בפורמט '\n'. שימוש ב-\n בלי מרכאות או בלי תווי escape יגרום לשימוש ב-"n" כתו מפריד.

    • ‫MINIMUM_OBJECT_CREATE_TIME: מציין את הזמן המינימלי שחלף מאז שאובייקט נוצר כדי שאפשר יהיה להוסיף אותו. הערך צריך להיות בפורמט UTC‏ YYYY-MM-DDThh:mm:ssZ. לדוגמה, 2024-10-14T08:30:30Z.

      כל תאריך, בעבר או בעתיד, מ-0001-01-01T00:00:00Z עד 9999-12-31T23:59:59Z כולל, הוא תקין.

    • ‫MATCH_GLOB: מציין את תבנית ה-glob להתאמה כדי שאובייקט ייקלט. כשמשתמשים ב-CLI של gcloud, אם יש ב-glob של ההתאמה תווים של *, צריך להוסיף לפני התו * את התו כדי לבטל את המשמעות המיוחדת שלו, כלומר להשתמש בפורמט \*\*.txt, או להוסיף את כל ה-glob של ההתאמה במירכאות "**.txt" או '**.txt'. מידע על התחביר הנתמך של תבניות glob מופיע במאמרי העזרה של Cloud Storage.

C++‎

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C++‎ במאמר תחילת העבודה: שימוש בספריות לקוח. מידע נוסף זמין במאמרי העזרה של Pub/Sub C++ API. 

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string bucket, std::string const& input_format,
   std::string text_delimiter, std::string match_glob,
   std::string const& minimum_object_create_time) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto& cloud_storage = *request.mutable_ingestion_data_source_settings()
                             ->mutable_cloud_storage();
  cloud_storage.set_bucket(std::move(bucket));
  if (input_format == "text") {
    cloud_storage.mutable_text_format()->set_delimiter(
        std::move(text_delimiter));
  } else if (input_format == "avro") {
    cloud_storage.mutable_avro_format();
  } else if (input_format == "pubsub_avro") {
    cloud_storage.mutable_pubsub_avro_format();
  } else {
    std::cout << "input_format must be in ('text', 'avro', 'pubsub_avro'); "
                 "got value: "
              << input_format << std::endl;
    return;
  }

  if (!match_glob.empty()) {
    cloud_storage.set_match_glob(std::move(match_glob));
  }

  if (!minimum_object_create_time.empty()) {
    google::protobuf::Timestamp timestamp;
    if (!google::protobuf::util::TimeUtil::FromString(
            minimum_object_create_time,
            cloud_storage.mutable_minimum_object_create_time())) {
      std::cout << "Invalid minimum object create time: "
                << minimum_object_create_time << std::endl;
    }
  }

  auto topic = client.CreateTopic(request);
  // Note that kAlreadyExists is a possible error when the library retries.
  if (topic.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The topic already exists\n";
    return;
  }
  if (!topic) throw std::move(topic).status();

  std::cout << "The topic was successfully created: " << topic->DebugString()
            << "\n";
}

המשך

בדוגמה הבאה נעשה שימוש בגרסה הראשית של ספריית הלקוח של Pub/Sub ב-Go (גרסה 2). אם אתם עדיין משתמשים בספרייה v1, כדאי לעיין במדריך להעברת נתונים ל-v2. כדי לראות רשימה של דוגמאות קוד מגרסה 1, אפשר לעיין ב דוגמאות הקוד שהוצאו משימוש.

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

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func createTopicWithCloudStorageIngestion(w io.Writer, projectID, topicID, bucket, matchGlob, minimumObjectCreateTime, delimiter string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	// bucket := "my-bucket"
	// matchGlob := "**.txt"
	// minimumObjectCreateTime := "2006-01-02T15:04:05Z"
	// delimiter := ","

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	minCreateTime, err := time.Parse(time.RFC3339, minimumObjectCreateTime)
	if err != nil {
		return err
	}

	topicpb := &pubsubpb.Topic{
		Name: fmt.Sprintf("projects/%s/topics/%s", projectID, topicID),
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_CloudStorage_{
				CloudStorage: &pubsubpb.IngestionDataSourceSettings_CloudStorage{
					Bucket: bucket,
					// Alternatively, can be Avro or PubSubAvro formats. See
					InputFormat: &pubsubpb.IngestionDataSourceSettings_CloudStorage_TextFormat_{
						TextFormat: &pubsubpb.IngestionDataSourceSettings_CloudStorage_TextFormat{
							Delimiter: &delimiter,
						},
					},
					MatchGlob:               matchGlob,
					MinimumObjectCreateTime: timestamppb.New(minCreateTime),
				},
			},
		},
	}
	t, err := client.TopicAdminClient.CreateTopic(ctx, topicpb)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Cloud storage topic created: %v\n", t)
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Java API. 


import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.protobuf.util.Timestamps;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.text.ParseException;

public class CreateTopicWithCloudStorageIngestionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    // Cloud Storage ingestion settings.
    // bucket and inputFormat are required arguments.
    String bucket = "your-bucket";
    String inputFormat = "text";
    String textDelimiter = "\n";
    String matchGlob = "**.txt";
    String minimumObjectCreateTime = "YYYY-MM-DDThh:mm:ssZ";

    createTopicWithCloudStorageIngestionExample(
        projectId, topicId, bucket, inputFormat, textDelimiter, matchGlob, minimumObjectCreateTime);
  }

  public static void createTopicWithCloudStorageIngestionExample(
      String projectId,
      String topicId,
      String bucket,
      String inputFormat,
      String textDelimiter,
      String matchGlob,
      String minimumObjectCreateTime)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      IngestionDataSourceSettings.CloudStorage.Builder cloudStorageBuilder =
          IngestionDataSourceSettings.CloudStorage.newBuilder().setBucket(bucket);
      switch (inputFormat) {
        case "text":
          cloudStorageBuilder.setTextFormat(
              IngestionDataSourceSettings.CloudStorage.TextFormat.newBuilder()
                  .setDelimiter(textDelimiter)
                  .build());
          break;
        case "avro":
          cloudStorageBuilder.setAvroFormat(
              IngestionDataSourceSettings.CloudStorage.AvroFormat.getDefaultInstance());
          break;
        case "pubsub_avro":
          cloudStorageBuilder.setPubsubAvroFormat(
              IngestionDataSourceSettings.CloudStorage.PubSubAvroFormat.getDefaultInstance());
          break;
        default:
          throw new IllegalArgumentException(
              "inputFormat must be in ('text', 'avro', 'pubsub_avro'); got value: " + inputFormat);
      }

      if (matchGlob != null && !matchGlob.isEmpty()) {
        cloudStorageBuilder.setMatchGlob(matchGlob);
      }

      if (minimumObjectCreateTime != null && !minimumObjectCreateTime.isEmpty()) {
        try {
          cloudStorageBuilder.setMinimumObjectCreateTime(Timestamps.parse(minimumObjectCreateTime));
        } catch (ParseException e) {
          System.err.println("Unable to parse timestamp: " + minimumObjectCreateTime);
        }
      }

      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder()
              .setCloudStorage(cloudStorageBuilder.build())
              .build();

      TopicName topicName = TopicName.of(projectId, topicId);

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println(
          "Created topic with Cloud Storage ingestion settings: " + topic.getAllFields());
    }
  }
}

Node.js

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במאמר תחילת העבודה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Node.js API. 

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const bucket = 'YOUR_BUCKET_NAME';
// const inputFormat = 'text';
// const textDelimiter = '\n';
// const matchGlob = '**.txt';
// const minimumObjectCreateTime = 'YYYY-MM-DDThh:mm:ssZ;

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithCloudStorageIngestion(
  topicNameOrId,
  bucket,
  inputFormat,
  textDelimiter,
  matchGlob,
  minimumObjectCreateTime,
) {
  const minimumDate = Date.parse(minimumObjectCreateTime);
  const topicMetadata = {
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      cloudStorage: {
        bucket,
        minimumObjectCreateTime: {
          seconds: minimumDate / 1000,
          nanos: (minimumDate % 1000) * 1000,
        },
        matchGlob,
      },
    },
  };

  // Make a format appropriately.
  switch (inputFormat) {
    case 'text':
      topicMetadata.ingestionDataSourceSettings.cloudStorage.textFormat = {
        delimiter: textDelimiter,
      };
      break;
    case 'avro':
      topicMetadata.ingestionDataSourceSettings.cloudStorage.avroFormat = {};
      break;
    case 'pubsub_avro':
      topicMetadata.ingestionDataSourceSettings.cloudStorage.pubsubAvroFormat =
        {};
      break;
    default:
      console.error('inputFormat must be in ("text", "avro", "pubsub_avro")');
      return;
  }

  // Creates a new topic with Cloud Storage ingestion.
  await pubSubClient.createTopic(topicMetadata);
  console.log(`Topic ${topicNameOrId} created with Cloud Storage ingestion.`);
}

Node.ts

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Node.js API. 

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const bucket = 'YOUR_BUCKET_NAME';
// const inputFormat = 'text';
// const textDelimiter = '\n';
// const matchGlob = '**.txt';
// const minimumObjectCreateTime = 'YYYY-MM-DDThh:mm:ssZ;

// Imports the Google Cloud client library
import {PubSub, TopicMetadata} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithCloudStorageIngestion(
  topicNameOrId: string,
  bucket: string,
  inputFormat: string,
  textDelimiter: string,
  matchGlob: string,
  minimumObjectCreateTime: string,
) {
  const minimumDate = Date.parse(minimumObjectCreateTime);
  const topicMetadata: TopicMetadata = {
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      cloudStorage: {
        bucket,
        minimumObjectCreateTime: {
          seconds: minimumDate / 1000,
          nanos: (minimumDate % 1000) * 1000,
        },
        matchGlob,
      },
    },
  };

  // Make a format appropriately.
  switch (inputFormat) {
    case 'text':
      topicMetadata.ingestionDataSourceSettings!.cloudStorage!.textFormat = {
        delimiter: textDelimiter,
      };
      break;
    case 'avro':
      topicMetadata.ingestionDataSourceSettings!.cloudStorage!.avroFormat = {};
      break;
    case 'pubsub_avro':
      topicMetadata.ingestionDataSourceSettings!.cloudStorage!.pubsubAvroFormat =
        {};
      break;
    default:
      console.error('inputFormat must be in ("text", "avro", "pubsub_avro")');
      return;
  }

  // Creates a new topic with Cloud Storage ingestion.
  await pubSubClient.createTopic(topicMetadata);
  console.log(`Topic ${topicNameOrId} created with Cloud Storage ingestion.`);
}

Python

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

from google.cloud import pubsub_v1
from google.protobuf import timestamp_pb2
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# bucket = "your-bucket"
# input_format = "text"  (can be one of "text", "avro", "pubsub_avro")
# text_delimiter = "\n"
# match_glob = "**.txt"
# minimum_object_create_time = "YYYY-MM-DDThh:mm:ssZ"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

cloud_storage_settings = IngestionDataSourceSettings.CloudStorage(
    bucket=bucket,
)
if input_format == "text":
    cloud_storage_settings.text_format = (
        IngestionDataSourceSettings.CloudStorage.TextFormat(
            delimiter=text_delimiter
        )
    )
elif input_format == "avro":
    cloud_storage_settings.avro_format = (
        IngestionDataSourceSettings.CloudStorage.AvroFormat()
    )
elif input_format == "pubsub_avro":
    cloud_storage_settings.pubsub_avro_format = (
        IngestionDataSourceSettings.CloudStorage.PubSubAvroFormat()
    )
else:
    print(
        "Invalid input_format: "
        + input_format
        + "; must be in ('text', 'avro', 'pubsub_avro')"
    )
    return

if match_glob:
    cloud_storage_settings.match_glob = match_glob

if minimum_object_create_time:
    try:
        minimum_object_create_time_timestamp = timestamp_pb2.Timestamp()
        minimum_object_create_time_timestamp.FromJsonString(
            minimum_object_create_time
        )
        cloud_storage_settings.minimum_object_create_time = (
            minimum_object_create_time_timestamp
        )
    except ValueError:
        print("Invalid minimum_object_create_time: " + minimum_object_create_time)
        return

request = Topic(
    name=topic_path,
    ingestion_data_source_settings=IngestionDataSourceSettings(
        cloud_storage=cloud_storage_settings,
    ),
)

topic = publisher.create_topic(request=request)

print(f"Created topic: {topic.name} with Cloud Storage Ingestion Settings")

נתקלתם בבעיות? היעזרו במאמר בנושא פתרון בעיות בייבוא של Cloud Storage.

עריכת נושא של ייבוא מ-Cloud Storage

אתם יכולים לערוך נושא ייבוא של Cloud Storage כדי לעדכן את המאפיינים שלו.

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

כדי לערוך נושא ייבוא של Cloud Storage, מבצעים את השלבים הבאים:

המסוף

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

    לדף Topics

  2. לוחצים על נושא הייבוא של Cloud Storage.

  3. בדף הפרטים של הנושא, לוחצים על עריכה.

  4. מעדכנים את השדות שרוצים לשנות.

  5. לוחצים על עדכון.

gcloud

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

    הפעלת Cloud Shell

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

  2. כדי לא לאבד את ההגדרות של נושא הייבוא, חשוב לכלול את כל ההגדרות בכל פעם שמעדכנים את הנושא. אם משאירים משהו בחוץ, Pub/Sub מאפס את ההגדרה לערך ברירת המחדל המקורי שלה.

    מריצים את הפקודה gcloud pubsub topics update עם כל הדגלים שמוזכרים בדוגמה הבאה:

    gcloud pubsub topics update TOPIC_ID \
    --cloud-storage-ingestion-bucket=BUCKET_NAME\
    --cloud-storage-ingestion-input-format=INPUT_FORMAT\
    --cloud-storage-ingestion-text-delimiter=TEXT_DELIMITER\
    --cloud-storage-ingestion-minimum-object-create-time=MINIMUM_OBJECT_CREATE_TIME\
    --cloud-storage-ingestion-match-glob=MATCH_GLOB

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

    • ‫TOPIC_ID הוא המזהה או השם של הנושא. אי אפשר לעדכן את השדה הזה.

    • ‫BUCKET_NAME: ציון השם של קטגוריה קיימת. לדוגמה, prod_bucket. שם הקטגוריה לא יכול לכלול את מזהה הפרויקט. במאמר יצירת קטגוריות מוסבר איך ליצור קטגוריה.

    • ‫INPUT_FORMAT: מציין את הפורמט של האובייקטים שמועברים. הערך יכול להיות text, avro או pubsub_avro. מידע נוסף על האפשרויות האלה מופיע במאמר בנושא פורמט קלט.

    • ‫TEXT_DELIMITER: מציין את התו שמגדיר את סוף השורה, שבאמצעותו יפוצלו אובייקטים של טקסט להודעות Pub/Sub. התו המפריד צריך להיות תו יחיד, וצריך להגדיר אותו רק אם הערך של INPUT_FORMAT הוא text. ברירת המחדל היא תו השורה החדשה (\n).

      כשמשתמשים ב-CLI של gcloud כדי לציין את התו שמפריד בין הערכים, צריך לשים לב במיוחד לטיפול בתווים מיוחדים כמו מעבר שורה \n. כדי לוודא שהמערכת תפרש את התו המפריד בצורה נכונה, צריך להשתמש בפורמט '\n'. אם משתמשים ב-\n בלי מירכאות או תווי בריחה, התוצאה היא תו מפריד של "n".

    • ‫MINIMUM_OBJECT_CREATE_TIME: מציין את הזמן המינימלי שחלף מאז שאובייקט נוצר כדי שאפשר יהיה להוסיף אותו. הערך צריך להיות בפורמט UTC‏ YYYY-MM-DDThh:mm:ssZ. לדוגמה, 2024-10-14T08:30:30Z.

      כל תאריך, בעבר או בעתיד, מ-0001-01-01T00:00:00Z עד 9999-12-31T23:59:59Z כולל, הוא תקין.

    • ‫MATCH_GLOB: מציין את תבנית ה-glob להתאמה כדי שאובייקט ייקלט. כשמשתמשים ב-CLI של gcloud, אם יש ב-glob של ההתאמה תווים של *, צריך להוסיף לפני התו * את התו כדי לבטל את המשמעות המיוחדת שלו, כלומר להשתמש בפורמט \*\*.txt , או להוסיף את כל ה-glob של ההתאמה במירכאות "**.txt" או '**.txt'. מידע על התחביר הנתמך של תבניות glob מופיע ב מסמכי Cloud Storage.

מכסות ומגבלות בנושא ייבוא ל-Cloud Storage

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

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