MongoDB-CDC

概述

MongoDB CDC 连接器允许从 MongoDB 读取快照数据和增量数据。本文档介绍如何设置 MongoDB CDC 连接器以对 MongoDB 运行 SQL 查询。

支持的版本

依赖项

I.为了设置 MongoDB CDC 连接器，下表提供了使用构建自动化工具（例如 Maven 或 SBT）的依赖关系信息

Maven依赖

<dependency>
    <groupId>org.apache.inlong</groupId>
    <artifactId>sort-connector-mongodb-cdc</artifactId>
    <version>1.2.0-incubating</version>
</dependency>

设置 MongoDB

可用性

• MongoDB 版本

  MongoDB 版本 >\= 3.6 我们使用 更改流功能（3.6 版中的新功能）来捕获更改数据。

• 集群部署

  需要 副本集或 分片集群。

• 存储引擎

  需要 WiredTiger存储引擎。

• 副本集协议版本

  需要副本集协议版本 1 (pv1)。 从版本 4.0 开始，MongoDB 仅支持 pv1。pv1 是使用 MongoDB 3.2 或更高版本创建的所有新副本集的默认值。

• 特权

  changeStream MongoDB Kafka 连接器 read 需要权限。

  您可以使用以下示例进行简单授权。 更详细的授权请参考 MongoDB 数据库用户角色。

  use admin;
  db.createUser({
    user: "flinkuser",
    pwd: "flinkpw",
    roles: [
      { role: "read", db: "admin" }, // read role includes changeStream privilege 
      { role: "readAnyDatabase", db: "admin" } // for snapshot reading
    ]
  });

如何创建 MongoDB Extract 节点

SQL API 用法

这个例子展示了如何使用 Flink SQL 创建一个 MongoDB Extract 节点:

-- Set checkpoint every 3000 milliseconds                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   
-- Create a MySQL table 'mongodb_extract_node' in Flink SQL
Flink SQL> CREATE TABLE mongodb_extract_node (
  _id STRING, // must be declared
  name STRING,
  weight DECIMAL(10,3),
  tags ARRAY<STRING>, -- array
  price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
  suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb-cdc',
  'hosts' = 'localhost:27017,localhost:27018,localhost:27019',
  'username' = 'flinkuser',
  'password' = 'flinkpw',
  'database' = 'inventory',
  'collection' = 'mongodb_extract_node'
);
-- Read snapshot and binlogs from mongodb_extract_node
Flink SQL> SELECT * FROM mongodb_extract_node;

注意

MongoDB 的更改事件记录在消息之前没有更新。所以，我们只能将其转换为 Flink 的 UPSERT 变更日志流。UPSERT 流需要唯一键，因此我们必须声明 _id 为主键。我们不能将其他列声明为主键，因为删除操作不包含除 _id 和 sharding key 之外的键和值。

InLong Dashboard 用法

TODO: 未来会支持

InLong Manager 用法

TODO: 未来会支持

MongoDB Extract 节点参数

可用元数据

以下格式元数据可以作为表定义中的只读 (VIRTUAL) 列公开。

扩展的 CREATE TABLE 示例演示了公开这些元数据字段的语法：

CREATE TABLE `mysql_extract_node` (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    collection_name STRING METADATA  FROM 'collection_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    _id STRING, // must be declared
    name STRING,
    weight DECIMAL(10,3),
    tags ARRAY<STRING>, -- array
    price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
    suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
    PRIMARY KEY(_id) NOT ENFORCED
) WITH (
      'connector' = 'mongodb-cdc', 
      'hostname' = 'YourHostname',
      'username' = 'YourUsername',
      'password' = 'YourPassword',
      'database' = 'YourDatabase',
      'collection' = 'YourTable' 
);

数据类型映射