ApsaraDB for MongoDB：使用MongoDB變更流（Change Stream）即時捕獲資料變更

監聽更多的DDL事件

前提條件

MongoDB 6.0及以上版本（升級指南）。

操作步驟

1. 使用 mongosh 串連資料庫。

2. 執行命令watch，設定 showExpandedEvents: true：

   // mongo shell or mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // 啟用 更多DDL 事件監聽
     }
   );
   cursor.next();

結果驗證

1. 在新的SQL視窗執行變更SQL，如db.createCollection("myCollection1")。

2. 觀察原SQL視窗輸出上述執行SQL相關資訊。

   

開啟前鏡像（Pre-image）

MongoDB的pre-image是指文檔在修改或刪除前的完整狀態快照，用於記錄資料變更前的原始值。

前提條件

MongoDB 6.0及以上版本（升級指南）。

操作步驟

1. 開啟資料庫維度前鏡像 ：

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" 表示使用oplog保留時間
       }
     }
   })

2. 開啟集合維度前鏡像 ：

   說明

   若需為某資料庫內所有集合啟用前鏡像，則必須先開啟資料庫維度前鏡像，然後在該庫的每個集合上單獨啟用集合級前鏡像功能。

   修改現有集合

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   建立新集合時指定

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. 建立變更流監聽 (指定Pre-image選項)：

   // 在目的地組合上建立監聽
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // 或 'whenAvailable'
       fullDocumentBeforeChange: 'required' // 或 'whenAvailable'
     }
   )
   cursor.next();

   • required: 服務端必須返回Pre/Post-image，否則報錯。

   • whenAvailable: 服務端儘力返回，不保證一定有。

結果驗證

1. 檢查資料庫維度開關是否已開啟：

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   開啟成功後輸出如下資訊：

2. 檢查集合配置：

   db.getCollectionInfos({name: "myCollection"}) // 或 db.runCommand({listCollections: 1})

   預期輸出：在返回的文檔中找到類似 "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } 的欄位。 

   

3. 在另一個mongosh視窗更新 myCollection 中的文檔。

4. 觀察 cursor 返回的事件，應包含 fullDocumentBeforeChange 欄位（變更前文檔）。

   

更多資訊參考Change Streams with Document Pre- and Post-Images。

開啟後鏡像（Post-image）

MongoDB的Post-image是變更發生後文檔的最新完整狀態快照，用於記錄變更後的完整文檔內容。

前提條件

MongoDB 3.6及以上版本（升級指南）。

操作步驟

執行watch命令時設定fullDocument: 'updateLookup'。

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();

結果驗證

1. 在另一個mongosh視窗新增或更新 myCollection 中的文檔。

2. 觀察 cursor 返回的事件，應包含 fullDocument 欄位（變更後文檔）。

處理超大變更事件 (>16MB)

前提條件

MongoDB 7.0及以上版本（升級指南）。

操作步驟 

在 watch() 的管道中包含 $changeStreamSplitLargeEvent 階段：

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // 添加切分階段
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

結果驗證

• 執行一個會導致變更事件 >16MB 的操作（如更新一個包含超大數組的文檔）。

• 觀察返回的事件流會被拆分成多個連續的fragment事件，最終以一個fragment事件結束。

降低前鏡像儲存開銷

預設Pre-image隨oplog到期。可設定更短的到期時間以節省空間的：

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // 單位：秒
    }
  }
})