קריאת נתונים בזמן אמת באמצעות סנכרון שינויים בזרמי נתונים

התכונה 'שינויים בסטרימינג' ב-Firestore עם תאימות ל-MongoDB מאפשרת לאפליקציות לגשת לשינויים בזמן אמת (הוספות, עדכונים ומחיקות) שבוצעו באוסף או במסד נתונים שלם. מקור נתונים לשינויים מסדר את העדכונים לפי זמן השינוי.

אפשר לגשת ל-Change Streams דרך ממשקי API שתואמים ל-MongoDB ודרייברים מסורתיים של MongoDB. ההטמעה של Change Streams ב-Firestore עם תאימות ל-MongoDB יכולה להתמודד עם כל נפח נתונים של פעולות כתיבה וקריאה באמצעות הטמעה ייחודית של חלוקה אוטומטית למחיצות בפעולות כתיבה וקריאה מקבילות. כך תוכלו ליצור עומסי עבודה עם תפוקה גבוהה. אפשר גם לשפר את התשתית להעברת נתונים ולסנכרון נתונים בין Cloud Firestore לבין פתרונות אחסון אחרים.

בנוסף לתאימות לדרייברים של MongoDB, אפשר להשתמש ב-Cloud Firestore כדי לקרוא Change Streams במקביל. כך אפשר ליצור עומסי עבודה מקבילים של קריאה עם תפוקה גבוהה. כל זרם מייצג חלוקה מאוזנת של התוצאות.

התכונה 'הזנת שינויים' תומכת בתכונות הבאות:

  • סנכרון שינויים בזרמי נתונים שניתנים להגדרה עם היקף מסד נתונים או אוסף.
  • משך השמירה של מקור נתונים לשינויים שמוגדר בזמן היצירה. ברירת המחדל של תקופת השמירה היא 7 ימים, ותקופת השמירה המינימלית היא יום אחד. השמירה צריכה להיות כפולה של יום אחד, עד 7 ימים לכל היותר. אי אפשר לשנות את משך השמירה אחרי שיוצרים את המדיניות. כדי לשנות את תקופת השמירה, צריך להפסיק את השימוש בפיד השינויים וליצור אותו מחדש.
  • אירועי השינוי delete, insert, update ו-drop ניתנים לצפייה באמצעות db.collection.watch() ו-db.watch().
  • ‫updateDescription.updatedFields מכיל הבדלים בין עדכונים.
  • כל האפשרויות של fullDocument ושל fullDocumentBeforeChange.
    • מחפש עדכונים במסמך המלא.
    • תמונה מקדימה של המסמך לפני שהוא הוחלף, עודכן או נמחק.
    • תמונה של המסמך אחרי שהוחלף או עודכן.
    • כדי לשחזר תמונות מלפני שעה או יותר, צריך להפעיל שחזור לנקודת זמן מסוימת (PITR).
  • כל אפשרויות החזרה למינוי, כולל resumeAfter ו-startAfter.
  • כשמשתמשים ב-watch() כדי לעקוב אחרי שינויים, אפשר לשרשר שלבי צבירה כמו $addFields,‏ $match,‏ $project,‏ $replaceRoot,‏ $replaceWith,‏ $set ו-$unset.

הגדרת סנכרון שינויים בזרמי נתונים

כדי ליצור, להסיר או להציג נתוני Change Streams קיימים במסד נתונים, משתמשים במסוף Google Cloud.

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

כדי ליצור, למחוק ולהציג ברשימה Change Streams, לחשבון המשתמש צריכות להיות הרשאות ניהול זהויות והרשאות גישה (IAM) מסוג datastore.schemas.create, datastore.schemas.delete ו-datastore.schemas.list, בהתאמה.

לדוגמה, התפקיד אדמין של אינדקסים במאגר נתונים (roles/datastore.indexAdmin) מעניק את ההרשאות האלה.

יצירת שידור חי של שינויים

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

כדי ליצור זרם שינויים, משתמשים במסוף Google Cloud.

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

    לדף Databases

  2. מהרשימה, בוחרים מסד נתונים של Firestore עם תאימות ל-MongoDB. החלונית Firestore Studio תיפתח.
  3. בחלונית Explorer (סייר), מחפשים את הצומת Change streams (סנכרון שינויים בזרמי נתונים), לוחצים על More actions (פעולות נוספות) ואז בוחרים באפשרות Create change stream (יצירת סנכרון שינויים בזרמי נתונים).
  4. מזינים שם ייחודי לשינוי הנתונים, היקף ותקופת שמירה, ואז לוחצים על שמירה.

הצגת שינויים בזרמי נתונים

אפשר לראות פרטים על Change Streams במסוף Google Cloud.

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

    לדף Databases

  2. מהרשימה, בוחרים מסד נתונים של Firestore עם תאימות ל-MongoDB. החלונית Firestore Studio תיפתח.
  3. בחלונית Explorer, מחפשים את הצומת סנכרון שינויים בזרמי נתונים.
  4. כדי לפתוח או לסגור את הצומת, לוחצים על החלפת צומת.

מחיקת שינוי השידור החי

כדי למחוק שינוי בנתונים, משתמשים במסוף Google Cloud.

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

    לדף Databases

  2. מהרשימה, בוחרים מסד נתונים של Firestore עם תאימות ל-MongoDB. החלונית Firestore Studio תיפתח.
  3. בחלונית Explorer, מחפשים את הצומת סנכרון שינויים בזרמי נתונים.
  4. כדי לפתוח או לסגור את הצומת, לוחצים על החלפת צומת.
  5. בסייר, מאתרים את הזרם של השינויים שרוצים למחוק.
  6. לוחצים על עוד פעולות ואז בוחרים באפשרות מחיקת נתוני השינויים.
  7. בתיבת הדו-שיח, מזינים את השם של זרם השינויים כדי לאשר את המחיקה, ואז לוחצים על מחיקה.

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

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

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

יצירת סמן של שינוי בנתונים

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

let cursor = db.watch()

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

let cursor = db.my_collection.watch()

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

let doc = db.my_collection.insertOne({value: "hello world"})
console.log(cursor.tryNext())

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

db.my_collection.updateOne({"_id": doc.insertedId}, {$set: {value: "hello world!"}})
db.my_collection.deleteOne({"_id": doc.insertedId}})

// Prints the update event
console.log(cursor.tryNext())

// Prints the delete event
console.log(cursor.tryNext())

המשך של שינוי בשידור

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

// Create a cursor and add one event to the change stream.
let cursor = db.my_collection.watch();
db.my_collection.insertOne({value: "hello world"});
let event = cursor.tryNext();

// Get the resume token from the event.
let resumeToken = event._id;

// Add a new event to the change stream.
db.my_collection.insertOne({value: "foobar"});

// Create a new cursor by using the resume token as a starting point.
let newCursor = db.my_collection.watch({resumeAfter: resumeToken})

// Log the change event containing the "foobar" value.
console.log(newCursor.tryNext())

כדי להשתמש ב-startAfter:

// Start after the resume token.
let startAfterCursor = db.my_collection.watch({startAfter: resumeToken})

הוספה של תמונות לפני ואחרי העדכונים ומחיקה שלהן

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

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

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

let cursor = db.my_collection.watch({
  "fullDocument": "required",
  "fullDocumentBeforeChange": "required"
})

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

במקום להחזיר שגיאה, אפשר להשתמש בערך whenAvailable כדי להחזיר את הערך null אם התמונות כבר לא זמינות.

let cursor = db.my_collection.watch({
  "fullDocument": "whenAvailable",
  "fullDocumentBeforeChange": "whenAvailable"
})

הכללת התמונה הנוכחית בעדכונים

כברירת מחדל, אירועי עדכון מכילים שדה updateDescription שהוא הדלתא של השדות ששונו על ידי פעולת העדכון. כדי לחפש את הגרסה העדכנית ביותר של המסמך כולו, משתמשים בערך updateLookup באפשרות fullDocument.

התכונה הזו לא דורשת PITR ומבצעת חיפוש של המסמך.

let cursor = db.my_collection.watch({
  "fullDocument": "updateLookup",
})

קריאות מקבילות

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

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

let cursor1 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 0 }}
  }]);

let cursor2 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 1 }}
  }]);

let cursor3 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 2 }}
  }]);

שינוי של זרמי נתונים וגיבויים

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

חיוב

הבדלים בהתנהגות

בקטע הבא מתוארים ההבדלים בזרמי השינויים בין Firestore עם תאימות ל-MongoDB לבין MongoDB.

updateDescription

‫updateDescription הוא מסמך באירוע update שמתאר את השדות שעודכנו או הוסרו על ידי פעולת העדכון. ב-Cloud Firestore, ההבדלים העיקריים הם:

  • ב-updateDescription, השדות truncatedArrays ו-disambiguatedPaths לא מאוכלסים.
  • ‫updateDescription.updatedFields מייצגות הבדל קנוני בין התמונות של מסמך לפני ואחרי החלת מוטציה.

נניח שמצב ההתחלתי של מסמך הוא:

db.my_collection.insertOne({
  _id: 1,
  root: {
    array: [{a: 1}, {b: 2}, {c: 3}]
  }
})

תרחיש 1: שינוי רק של הרכיב הראשון במערך.

בתרחיש הזה, ההתנהגות של Cloud Firestore זהה לזו של MongoDB.

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array.0.a": 100}}
)

{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

תרחיש 2: החלפה של מערך שלם

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

ההפרש בעדכון של Cloud Firestore לא מבדיל בין שני התרחישים האלה ומחזיר את אותו updateDescription.updatedFields בשניהם:

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array": [{a: 100}, {b: 2}, {c: 3}]}}
)

// In other implementations, updatedFields reflects the mutation itself
{
  updatedFields: {
    "root.array": [{a: 100}, {b: 2}, {c: 3}]
  },
  removedFields: []
}

// Firestore updatedFields is the diff between the before and after versions of the document
{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

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