Realtime Compute for Apache Flink：MongoDB

WITH參數

• 通用

  參數	說明	資料類型	是否必填	預設值	備忘
  connector	連接器名稱。	String	是	無	作為源表： Realtime Compute引擎VVR 8.0.4及之前版本，填寫為mongodb-cdc。 Realtime Compute引擎VVR 8.0.5及之後版本，填寫為mongodb或mongodb-cdc。 作為維表或結果表時，固定值為mongodb。
  uri	MongoDB串連uri。	String	否	無	說明 參數uri與hosts必須指定其中之一。若指定uri，則無需指定scheme、hosts、username、password、connector.options。當兩者均指定時將使用uri進行串連。
  hosts	MongoDB所在的主機名稱。	String	否	無	可以使用英文逗號（,）分隔提供多個主機名稱。
  scheme	MongoDB使用的連線協定。	String	否	mongodb	可選的取值包括： mongodb：代表使用預設的MongoDB協議進行串連 mongodb+srv：代表使用DNS SRV記錄協議進行串連
  username	串連到MongoDB時使用的使用者名稱。	String	否	無	開啟身分識別驗證功能時，必須配置該參數。
  password	串連到MongoDB時使用的密碼。	String	否	無	開啟身分識別驗證功能時，必須配置該參數。 重要 為了避免您的密碼資訊泄露，建議您通過密鑰管理的方式填寫密碼取值，詳情請參見變數管理。
  database	MongoDB資料庫名稱。	String	否	無	作為源表時，資料庫名稱支援Regex匹配。 不配置該參數代表監控全部資料庫。
  collection	MongoDB集合名稱。	String	否	無	作為源表時，集合名稱支援Regex匹配。 重要 如果您要監控的集合名稱中包含Regex特殊字元，則必須提供完整名字空間（資料庫名稱.集合名稱），否則無法捕獲對應集合的變更。 不配置該參數代表監控全部集合。
  connection.options	MongoDB側的串連參數。	String	否	無	使用&分隔的key=value式額外串連參數。例如connectTimeoutMS=12000&socketTimeoutMS=13000。

• 源表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  scan.startup.mode	MongoDB CDC的啟動模式。	String	否	initial	參數取值如下： initial：從初始位點開始拉取全部資料。 latest-offset：從當前位點開始拉取變更資料。 timestamp：從指定的時間戳記開始拉取變更資料。 詳情請參見Startup Properties。
  scan.startup.timestamp-millis	指錨點消費的起始時間戳記。	Long	取決於 scan.startup.mode的取值 initial：否 latest-offset：否 timestamp：是	無	參數格式為自Linux Epoch時間戳記以來的毫秒數。 僅適用於timestamp啟動模式。
  initial.snapshotting.queue.size	進行初始快照集時的隊列大小限制。	Integer	否	10240	僅在scan.startup.mode選項設定為initial 時生效。
  batch.size	遊標的批處理大小。	Integer	否	1024	無。
  poll.max.batch.size	同一批處理的最多變更文檔數量。	Integer	否	1024	此參數控制流程處理時一次拉取最多變更文檔的個數。取值越大，連接器內部分配的緩衝區越大。
  poll.await.time.ms	兩次拉取資料之間的時間間隔。	Integer	否	1000	單位為毫秒。
  heartbeat.interval.ms	發送心跳包的時間間隔。	Integer	否	0	單位為毫秒。 MongoDB CDC連接器主動向資料庫發送心跳包來保證回溯狀態最新。設定為0代表永不發送心跳包。 重要 對於更新不頻繁的集合，強烈建議設定此選項。
  scan.incremental.snapshot.enabled	是否啟用並行模式進行初始快照集。	Boolean	否	false	實驗性功能。
  scan.incremental.snapshot.chunk.size.mb	並行模式讀取快照時的分區大小。	Integer	否	64	實驗性功能。 單位為MB。 僅在啟用並行快照時生效。
  scan.full-changelog	產生完整的Full Changelog事件流。	Boolean	否	false	實驗性功能。 說明 MongoDB資料庫需要為6.0及以上版本，並且已開啟前像後像功能，開啟方法請參見Document Preimages。
  scan.flatten-nested-columns.enabled	是否將以.分隔的欄位名解析為嵌套BSON文檔讀取。	Boolean	否	false	若開啟，在如下樣本的BSON文檔中，col欄位在schema中名稱為nested.col。 {"nested":{"col":true}} 說明 僅VVR 8.0.5及以上版本支援該參數。
  scan.primitive-as-string	是否將BSON文檔中的原始類型都解析為字串類型。	Boolean	否	false	說明 僅VVR 8.0.5及以上版本支援該參數。

• 維表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  lookup.cache	Cache策略。	String	否	NONE	目前支援以下兩種緩衝策略： None：無緩衝。 Partial：只在外部資料庫中尋找資料時緩衝。
  lookup.max-retries	查詢資料庫失敗的最大重試次數。	Integer	否	3	無。
  lookup.retry.interval	如果查詢資料庫失敗，重試的時間間隔。	Duration	否	1s	無。
  lookup.partial-cache.expire-after-access	緩衝中的記錄最長保留時間。	Duration	否	無	支援時間單位ms、s、min、h和d。 使用該配置時 lookup.cache 必須設定為 PARTIAL。
  lookup.partial-cache.expire-after-write	在記錄寫入緩衝後該記錄的最大保留時間。	Duration	否	無	使用該配置時 lookup.cache 必須設定為 PARTIAL。
  lookup.partial-cache.max-rows	緩衝的最大條數。超過該值，最舊的行將到期。	Long	否	無	使用該配置時 lookup.cache 必須設定為 PARTIAL。
  lookup.partial-cache.cache-missing-key	在物理表中未關聯到資料時，是否緩衝空記錄。	Boolean	否	True	使用該配置時 lookup.cache 必須設定為 PARTIAL。

• 結果表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  sink.buffer-flush.max-rows	每次按批寫入資料時的最大記錄數。	Integer	否	1000	無。
  sink.buffer-flush.interval	寫入資料的重新整理間隔。	Duration	否	1s	無。
  sink.delivery-guarantee	寫入資料時的語義保證。	String	否	at-least-once	可選的取值包括： none at-least-once 說明 目前不支援exactly-once。
  sink.max-retries	寫入資料庫失敗時的最大重試次數。	Integer	否	3	無。
  sink.retry.interval	寫入資料庫失敗時的重試時間間隔。	Duration	否	1s	無。
  sink.parallelism	自訂sink並行度。	Integer	否	空	無。

類型映射

• CDC源表

  BSON類型	Flink SQL類型
  Int32	INT
  Int64	BIGINT
  Double	DOUBLE
  Decimal128	DECIMAL(p, s)
  Boolean	BOOLEAN
  Date Timestamp	DATE
  Date Timestamp	TIME
  DateTime	TIMESTAMP(3) TIMESTAMP_LTZ(3)
  Timestamp	TIMESTAMP(0) TIMESTAMP_LTZ(0)
  String ObjectId UUID Symbol MD5 JavaScript Regex	STRING
  Binary	BYTES
  Object	ROW
  Array	ARRAY
  DBPointer	ROW<$ref STRING, $id STRING>
  GeoJSON	Point: ROW<type STRING, coordinates ARRAY<DOUBLE>> Line: ROW<type STRING, coordinates ARRAY<ARRAY< DOUBLE>>>

• 維表和結果表

  BSON類型	Flink SQL類型
  Int32	INT
  Int64	BIGINT
  Double	DOUBLE
  Decimal128	DECIMAL
  Boolean	BOOLEAN
  DateTime	TIMESTAMP_LTZ(3)
  Timestamp	TIMESTAMP_LTZ(0)
  String ObjectId	STRING
  Binary	BYTES
  Object	ROW
  Array	ARRAY

使用樣本

• CDC源表

  CREATE TEMPORARY TABLE mongo_source (
    `_id` STRING, --must be declared
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price ROW<amount DECIMAL, currency STRING>,
    suppliers ARRAY<ROW<name STRING, address STRING>>,
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    collection_name STRING METADATA VIRTUAL,
    op_ts TIMESTAMP_LTZ(3) METADATA VIRTUAL,
    PRIMARY KEY(_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mongodb',
    'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
    'username' = 'root',
    'password' = '${secret_values.password}',
    'database' = 'flinktest',
    'collection' = 'flinkcollection',
    'scan.incremental.snapshot.enabled' = 'true',
    'scan.full-changelog' = 'true'
  );
  
  CREATE TEMPORARY TABLE  productssink (
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price_amount DECIMAL,
    suppliers_name STRING,
    db_name STRING,
    collection_name STRING,
    op_ts TIMESTAMP_LTZ(3)
  ) WITH (
    'connector' = 'print',
    'logger' = 'true'
  );
  
  INSERT INTO productssink  
  SELECT
    name,
    weight,
    tags,
    price.amount,
    suppliers[1].name,
    db_name,
    collection_name,
    op_ts
  FROM
    mongo_source;

• 維表

  CREATE TEMPORARY TABLE datagen_source (
    id STRING,
    a int,
    b BIGINT,
    `proctime` AS PROCTIME()
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mongo_dim (
    `_id` STRING,
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price ROW<amount DECIMAL, currency STRING>,
    suppliers ARRAY<ROW<name STRING, address STRING>>,
    PRIMARY KEY(_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mongodb',
    'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
    'username' = 'root',
    'password' = '${secret_values.password}',
    'database' = 'flinktest',
    'collection' = 'flinkcollection',
    'lookup.cache' = 'PARTIAL',
    'lookup.partial-cache.expire-after-access' = '10min',
    'lookup.partial-cache.expire-after-write' = '10min',
    'lookup.partial-cache.max-rows' = '100'
  );
  
  CREATE TEMPORARY TABLE print_sink (
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price_amount DECIMAL,
    suppliers_name STRING
  ) WITH (
    'connector' = 'print',
    'logger' = 'true'
  );
  
  INSERT INTO print_sink
  SELECT
    T.id,
    T.a,
    T.b,
    H.name
  FROM
    datagen_source AS T JOIN mongo_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.id = H._id;

• 結果表

  CREATE TEMPORARY TABLE datagen_source (
    `_id` STRING,
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price ROW<amount DECIMAL, currency STRING>,
    suppliers ARRAY<ROW<name STRING, address STRING>>
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mongo_sink (
    `_id` STRING,
    name STRING,
    weight DECIMAL,
    tags ARRAY<STRING>,
    price ROW<amount DECIMAL, currency STRING>,
    suppliers ARRAY<ROW<name STRING, address STRING>>,
    PRIMARY KEY(_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mongodb',
    'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
    'username' = 'root',
    'password' = '${secret_values.password}',
    'database' = 'flinktest',
    'collection' = 'flinkcollection'
  );
  
  INSERT INTO mongo_sink
  SELECT * FROM datagen_source;

中繼資料

MongoDB CDC源表支援中繼資料列文法，您可以通過中繼資料列訪問以下中繼資料。

關於MongoDB的變更前後像記錄功能

MongoDB 6.0 之前的版本預設不會提供變更前文檔及被刪除文檔的資料，在未開啟變更前後像記錄功能時，利用已有資訊只能實現 Upsert 語義（即缺失了 Update Before 資料條目）。但在 Flink 中許多有用的運算元操作都依賴完整的 Insert、Update Before、Update After、Delete 變更流。

為了補充缺失的變更前事件，目前 Flink SQL Planner 會自動為 Upsert 類型的資料來源產生一個 ChangelogNormalize 節點，該節點會在 Flink 狀態中緩衝所有文檔的目前的版本快照，在遇到被更新或刪除的文檔時，查表即可得知變更前的狀態，但該運算元節點需要儲存體積巨大的狀態資料。

MongoDB 6.0版本支援開啟資料庫的前像後像（Pre- and Post-images）記錄功能，詳情可參考Document Preimages。開啟該功能後，MongoDB會在每次變更發生時，在一個特殊的集合中記錄文檔變更前後的完整狀態。此時在作業中啟用scan.full-changelog配置項，MongoDB CDC會從變更文檔記錄中產生Update Before記錄，從而支援產生完整事件流，消除了對ChangelogNormalize節點的依賴。

Mongo CDC DataStream API

建立DataStream API程式並使用MongoDBSource。程式碼範例如下：

MongoDBSource.builder()
  .hosts("mongo.example.com:27017")
  .username("mongouser")
  .password("mongopasswd")
  .databaseList("testdb")
  .collectionList("testcoll")
  .startupOptions(StartupOptions.initial())
  .deserializer(new JsonDebeziumDeserializationSchema())
  .build();

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>flink-connector-mongodb</artifactId>
    <version>${vvr.version}</version>
</dependency>

在構造MongoDBSource時，可以配置以下參數：