MongoDB（公测中）

WITH参数

• 通用

  参数	说明	数据类型	是否必填	默认值	备注
  connector	连接器名称。	String	是	无	作为源表： 实时计算引擎VVR 8.0.4及之前版本，填写为mongodb-cdc。 实时计算引擎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	否	无	作为源表时，数据库名称支持正则表达式匹配。 不配置该参数代表监控全部数据库。
  collection	MongoDB集合名称。	String	否	无	作为源表时，集合名称支持正则表达式匹配。 重要 如果您要监控的集合名称中包含正则表达式特殊字符，则必须提供完全限定的名字空间（数据库名称.集合名称），否则无法捕获对应集合的变更。 不配置该参数代表监控全部集合。
  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 After记录，从而支持产生完整事件流，消除了对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();

在构造MongoDBSource时，可以配置以下参数：