อ่านข้อมูลแบบเรียลไทม์ด้วยสตรีมการเปลี่ยนแปลง

Change Streams สำหรับ Firestore ที่มีความเข้ากันได้กับ MongoDB ช่วยให้แอปพลิเคชันเข้าถึงการเปลี่ยนแปลงแบบเรียลไทม์ (การแทรก การอัปเดต และการลบ) ที่ทำกับคอลเล็กชันหรือฐานข้อมูลทั้งหมดได้ สตรีมการเปลี่ยนแปลงจะจัดเรียงการอัปเดตตามเวลาที่แก้ไข

Change Streams เข้าถึงได้ผ่าน API ที่เข้ากันได้กับ MongoDB และไดรเวอร์ MongoDB แบบดั้งเดิม การใช้งาน Change Streams ของ Firestore ที่เข้ากันได้กับ MongoDB สามารถจัดการอัตราการส่งข้อมูลของการเขียนและการอ่านผ่านการใช้งานเฉพาะของการแบ่งพาร์ติชันอัตโนมัติในการเขียนและการอ่านแบบคู่ขนาน ซึ่งจะช่วยให้คุณสร้างภาระงานที่มีอัตราการส่งผ่านข้อมูลสูงได้ นอกจากนี้ คุณยังปรับปรุงโครงสร้างพื้นฐานการย้ายข้อมูลและการซิงค์ข้อมูลระหว่าง Cloud Firestore กับโซลูชันพื้นที่เก็บข้อมูลอื่นๆ ได้ด้วย

นอกเหนือจากความเข้ากันได้กับไดรเวอร์ MongoDB แล้ว คุณยังใช้ Cloud Firestore เพื่ออ่านสตรีมการเปลี่ยนแปลงแบบคู่ขนานได้ด้วย ซึ่งช่วยให้คุณสร้างภาระงานการอ่านแบบคู่ขนานที่มีอัตราการส่งผ่านข้อมูลสูงได้ แต่ละสตรีมแสดงถึงพาร์ติชันของผลลัพธ์ที่มีการกระจายอย่างเหมาะสม

สตรีมการเปลี่ยนแปลงรองรับฟีเจอร์ต่อไปนี้

  • สตรีมการเปลี่ยนแปลงที่กำหนดค่าได้ซึ่งมีขอบเขตฐานข้อมูลหรือคอลเล็กชัน
  • ระยะเวลาเก็บรักษาสำหรับสตรีมการเปลี่ยนแปลงที่ระบุไว้เมื่อสร้าง ระยะเวลาเก็บรักษาเริ่มต้นคือ 7 วัน และระยะเวลาเก็บรักษาขั้นต่ำคือ 1 วัน ระยะเวลาเก็บรักษาต้องเป็นจำนวนเท่าของ 1 วัน โดยสูงสุดไม่เกิน 7 วัน ระยะเวลาเก็บรักษาจะเปลี่ยนแปลงไม่ได้หลังจากที่สร้างแล้ว หากต้องการเปลี่ยนระยะเวลาเก็บรักษา คุณต้องวางและสร้างสตรีมการเปลี่ยนแปลงอีกครั้ง
  • เหตุการณ์การเปลี่ยนแปลง delete, insert, update และ drop ที่สังเกตได้โดยใช้ db.collection.watch() และ db.watch()
  • updateDescription.updatedFields มีความแตกต่างของการอัปเดต
  • ตัวเลือก fullDocument และ fullDocumentBeforeChange ทั้งหมด
    • กำลังค้นหาเอกสารฉบับเต็มเพื่อดูข้อมูลอัปเดต
    • รูปภาพก่อนหน้าของเอกสารก่อนที่จะมีการแทนที่ อัปเดต หรือลบ
    • ภาพหลังการดำเนินการของเอกสารหลังจากที่มีการแทนที่หรืออัปเดต
    • รูปภาพก่อนและหลังที่เก่ากว่า 1 ชั่วโมงต้องเปิดใช้การกู้คืนช่วงเวลา (PITR)
  • ตัวเลือกการกลับมาทำงานต่อทั้งหมด รวมถึง resumeAfter และ startAfter
  • เมื่อใช้ watch() เพื่อสังเกตการเปลี่ยนแปลง คุณสามารถเชื่อมโยงระยะการรวม เช่น $addFields, $match, $project, $replaceRoot, $replaceWith, $set และ $unset

กำหนดค่าสตรีมการเปลี่ยนแปลง

หากต้องการสร้าง วาง หรือดูสตรีมการเปลี่ยนแปลงที่มีอยู่สำหรับฐานข้อมูล ให้ใช้คอนโซล Google Cloud

บทบาทและสิทธิ์

หากต้องการสร้าง ลบ และแสดงรายการสตรีมการเปลี่ยนแปลง บุคคลหลักต้องมีสิทธิ์ Identity and Access Management (IAM) datastore.schemas.create, datastore.schemas.delete และ datastore.schemas.list ตามลำดับ

ตัวอย่างเช่น บทบาทผู้ดูแลระบบดัชนีพื้นที่เก็บข้อมูล (roles/datastore.indexAdmin) จะให้สิทธิ์เหล่านี้

สร้างสตรีมการเปลี่ยนแปลง

คุณต้องสร้าง Change Stream ก่อนจึงจะเปิดเคอร์เซอร์ Change Stream ที่เกี่ยวข้องได้ ไม่รองรับการเปิดใช้สตรีมการเปลี่ยนแปลงอัตโนมัติที่คอลเล็กชันหรือการสร้างฐานข้อมูล

หากต้องการสร้างสตรีมการเปลี่ยนแปลง ให้ใช้คอนโซล Google Cloud

  1. ในคอนโซล Google Cloud ให้ไปที่หน้าฐานข้อมูล

    ไปที่ฐานข้อมูล

  2. จากรายการ ให้เลือกฐานข้อมูล Firestore ที่มีความเข้ากันได้กับ MongoDB แผง Firestore Studio จะเปิดขึ้น
  3. ในแผง Explorer ให้ค้นหาโหนดสตรีมการเปลี่ยนแปลง คลิก การดำเนินการเพิ่มเติม แล้วเลือกสร้างสตรีมการเปลี่ยนแปลง
  4. ป้อนชื่อสตรีมการเปลี่ยนแปลง ขอบเขต และระยะเวลาเก็บรักษาที่ไม่ซ้ำกัน แล้วคลิกบันทึก

ดูสตรีมการเปลี่ยนแปลง

คุณดูรายละเอียดเกี่ยวกับสตรีมการเปลี่ยนแปลงได้ในคอนโซล Google Cloud

  1. ในคอนโซล Google Cloud ให้ไปที่หน้าฐานข้อมูล

    ไปที่ฐานข้อมูล

  2. จากรายการ ให้เลือกฐานข้อมูล Firestore ที่มีความเข้ากันได้กับ MongoDB แผง Firestore Studio จะเปิดขึ้น
  3. ในแผง Explorer ให้ค้นหาโหนดสตรีมการเปลี่ยนแปลง
  4. หากต้องการเปิดหรือปิดโหนด ให้คลิก สลับโหนด

ลบสตรีมการเปลี่ยนแปลง

หากต้องการลบสตรีมการเปลี่ยนแปลง ให้ใช้คอนโซล Google Cloud

  1. ในคอนโซล Google Cloud ให้ไปที่หน้าฐานข้อมูล

    ไปที่ฐานข้อมูล

  2. จากรายการ ให้เลือกฐานข้อมูล Firestore ที่มีความเข้ากันได้กับ MongoDB แผง Firestore Studio จะเปิดขึ้น
  3. ในแผง Explorer ให้ค้นหาโหนดสตรีมการเปลี่ยนแปลง
  4. หากต้องการเปิดหรือปิดโหนด ให้คลิก สลับโหนด
  5. ใน Explorer ให้ค้นหา Change Stream ที่ต้องการลบ
  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) และหากต้องการอ่านรูปภาพเอกสารที่มีอายุมากกว่า 1 ชั่วโมง คุณต้องเปิดใช้ 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 ไม่ได้แยกความแตกต่างระหว่าง 2 สถานการณ์นี้ และแสดงผล updateDescription.updatedFields เดียวกันสำหรับทั้ง 2 สถานการณ์ ดังนี้

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: []
}

ขั้นตอนถัดไป