[Unix] Flink CDC 系列 - Flink MongoDB CDC 在 XTransfer 的生产实践

• ​INSERT_ONLY：​只包含 +I，适用于批处理和事件流。
  
• ​ALL：​包含 +I, -U, +U, -D 全部的 RowKind，如 MySQL binlog。
  
• UPSERT：只包含 +I, +U, -D 三种类型的 RowKind，不包含 -U，但必须按唯一键的幂等更新 , 如 MongoDB Change Streams。
  

{
  "ts" : Timestamp(1640190995, 3),
  "t" : NumberLong(434),
  "h" : NumberLong(3953156019015894279),
  "v" : 2,
  "op" : "u",
  "ns" : "db.firm",
  "ui" : UUID("19c72da0-2fa0-40a4-b000-83e038cd2c01"),
  "o2" : {
    "_id" : ObjectId("61c35441418152715fc3fcbc")
  },
  "wall" : ISODate("2021-12-22T16:36:35.165Z"),
  "o" : {
    "$v" : 1,
    "$set" : {
      "address" : "Shanghai China"
    }
  }
}

• WiredTiger 存储引擎
  
• 副本集 (测试环境下，也可以使用单节点的副本集) 或分片集群部署
  
• 副本集协议版本：pv1 (默认)
  
• 4.0 版本之前允许 Majority Read Concern: replication.enableMajorityReadConcern = true (默认允许)
  
• MongoDB 用户拥有 find 和 changeStream 权限
  

{
   _id : { <BSON Object> },
   "operationType" : "<operation>",
   "fullDocument" : { <document> },
   "ns" : {
    "db" : "<database>",
    "coll" : "<collection>"
   },
   "to" : {
    "db" : "<database>",
    "coll" : "<collection>"
   },
   "documentKey" : { "_id" : <value> },
   "updateDescription" : {
    "updatedFields" : { <document> },
    "removedFields" : [ "<field>", ... ],
    "truncatedArrays" : [
     { "field" : <field>, "newSize" : <integer> },
     ...
    ]
   },
   "clusterTime" : <Timestamp>,
   "txnNumber" : <NumberLong>,
   "lsid" : {
    "id" : <UUID>,
    "uid" : <BinData>
   }
}

字段类型描述_iddocument表示resumeTokenoperationTypestring操作类型，包括：insert, delete, replace, update, drop, rename, dropDatabase, invalidatefullDocumentdocument完整文档记录，insert, replace默认包含，update需要开启updateLookup，delete和其他操作类型不包含nsdocument操作记录对应集合的完全名称todocument当操作类型为rename时，to表示重命名后的完全名称documentKeydocument包含变更文档的主键 _id，如果该集合是一个分片集合，documentKey中也会包含分片建updateDescriptiondocument当操作类型为update时，描述有变更的字段和值clusterTimeTimestamp操作时间txnNumberNumberLong事务号lsidDocumentsession id

• 支持 Exactly-Once 语义
  
• 支持全量、增量订阅
  
• 支持 Snapshot 数据过滤
  
• 支持从检查点、保存点恢复
  
• 支持元数据提取
  

db.adminCommand(
  {
  replSetResizeOplog: 1, // 固定值1
  size: 20480,       // 单位为MB，范围在990MB到1PB
  minRetentionHours: 168 // 可选项，单位为小时
  }
)

WITH (
  'connector' = 'mongodb-cdc',
  'heartbeat.interval.ms' = '60000'
)

WITH (
   'connector' = 'mongodb-cdc',
   'connection.options' = 'authSource=authDB&maxPoolSize=3'
)

• poll.await.time.ms
  

• poll.max.batch.size
  

MongoDBSource.<String>builder()
  .hosts("127.0.0.1:27017")
  .database("")
  .collection("")
  .pipeline("[{'$match': {'ns.db': {'$regex': '/^(sandbox|firewall)$/'}}}]")
  .deserializer(new JsonDebeziumDeserializationSchema())
  .build();

• 单集合
  

{ resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }

• 单库
  

{ resource: { db: <dbname>, collection: "" }, actions: [ "find", "changeStream" ] }

• 集群
  

{ resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

use admin;
// 创建用户
db.createUser(
 {
   user: "flink",
   pwd: "flinkpw",
   roles: []
 }
);
// 创建角色
db.createRole(
   {
   role: "flink_role", 
   privileges: [
     { resource: { db: "inventory", collection: "products" }, actions: [ "find", "changeStream" ] }
   ],
   roles: []
   }
);
// 给用户授予角色
db.grantRolesToUser(
  "flink",
  [
    // 注意：这里的db指角色创建时的db，在admin下创建的角色可以包含不同database的访问权限
    { role: "flink_role", db: "admin" }
  ]
);
// 给角色追加权限
db.grantPrivilegesToRole(
  "flink_role",
   [
     { resource: { db: "inventory", collection: "orders" }, actions: [ "find", "changeStream" ] }
   ]
);

use admin;
db.createUser({
  user: "flink",
  pwd: "flinkpw",
  roles: [
  { role: "read", db: "admin" },
  { role: "readAnyDatabase", db: "admin" }
  ]
});

• ​支持增量 Snapshot​
  目前，MongoDB CDC Connector 还不支持增量 Snapshot，对于数据量较大的表还不能很好发挥 Flink 并行计算的优势。后续将实现 MongoDB 的增量 Snapshot 功能，使其支持 Snapshot 阶段的 checkpoint，和并发度设置。
  

• 支持从指定时间进行变更订阅
  目前，MongoDB CDC Connector 仅支持从当前时间开始 Change Stream 的订阅，后续将提供从指定时间点的 Change Stream 订阅。
  

• ​支持库和集合的筛选​
  目前，MongoDB CDC Connector 支持集群、整库的变更订阅和筛选，但对于是否需要进行 Snapshot 的集合的筛选还不支持，后续将完善这个功能。
  

跨境支付独角兽企业XTransfer在招技术/产品运营类岗位，包括系统架构师、开发工程师、算法工程师、测试工程师、产品经理、UI/交互设计师等数十个人员。产研方向的同学简历直达渠道：wenjuan.he@xtransfer.cn