MongoDB SQL 连接器

Scan Source: Bounded Lookup Source: Sync Mode Sink: Batch Sink: Streaming Append & Upsert Mode

MongoDB 连接器提供了从 MongoDB 中读取和写入数据的能力。 本文档介绍如何设置 MongoDB 连接器以对 MongoDB 运行 SQL 查询。

连接器可以在 upsert 模式下运行，使用在 DDL 上定义的主键与外部系统交换 UPDATE/DELETE 消息。

如果 DDL 上没有定义主键，则连接器只能以 append 模式与外部系统交换 INSERT 消息且不支持消费 UPDATE/DELETE 消息。

依赖

In order to use the MongoDB connector the following dependencies are required for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-mongodb</artifactId>
      <version>1.1.0-1.18</version>
    </dependency>

MongoDB 连接器目前并不包含在 Flink 的二进制发行版中，请查阅这里了解如何在集群运行中引用 MongoDB 连接器。

如何创建 MongoDB 表

MongoDB 表可以以如下方式定义:

-- 在 Flink SQL 中注册一张 MongoDB 'users' 表
CREATE TABLE MyUserTable (
  _id STRING,
  name STRING,
  age INT,
  status BOOLEAN,
  PRIMARY KEY (_id) NOT ENFORCED
) WITH (
   'connector' = 'mongodb',
   'uri' = 'mongodb://user:password@127.0.0.1:27017',
   'database' = 'my_db',
   'collection' = 'users'
);
-- 读取表 "T" 的数据并写入至 MongoDB 表
INSERT INTO MyUserTable
SELECT _id, name, age, status FROM T;
-- 从 MongoDB 表中读取数据
SELECT id, name, age, status FROM MyUserTable;
-- 将 MongoDB 表作为维度表关联
SELECT * FROM myTopic
LEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctime
ON myTopic.key = MyUserTable._id;

连接器参数

特性

键处理

当写入数据到外部数据库时，Flink 会使用 DDL 中定义的主键。 如果定义了主键，则连接器将以 upsert 模式工作，否则连接器将以 append 模式工作。

在 MongoDB 连接器中主键用来计算 MongoDB 文档的保留主键 _id。 MongoDB 的文档 _id 是唯一的且不可变的，可以是除了 Array 外的任何 BSON 类型。 如果 _id 是嵌套的，嵌套字段的名称不能包含 ($) 符号。

MongoDB 的主键有一些限制。 在 MongoDB 4.2 之前，索引值的大小需要小于 1024 字节。 在 MongoDB 4.2 及之后版本，移除了索引大小的限制。 请参阅 Index Key Limit 了解更多详情。

MongoDB 连接器通过将 DDL 声明的主键进行组合，来生成文档的保留主键 _id。

• 当 DDL 中声明的主键仅包含一个字段时，将字段数据转换为 bson 类型的值作为相应文档的_id。
• 当 DDL 中声明的主键包含多个字段时，我们将所有声明的主键字段组合为一个 bson document 的嵌套类型，作为相应文档的_id。 举个例子，如果我们在 DDL 中定义主键为 PRIMARY KEY (f1, f2) NOT ENFORCED，那么相应文档的 _id 将会是 _id: {f1: v1, f2: v2} 的形式。

需要注意的是：当 _id 字段存在于 DDL 中，但是主键并没有指定为 _id 时，会产生歧义。 要么将 _id 字段声明为主键，或者将 _id 字段进行重命名。

有关 PRIMARY KEY 语法的更多详细信息，请参见 CREATE TABLE DDL。

分区扫描

为了在并行 Source task 实例中加速读取数据，Flink 为 MongoDB table 提供了分区扫描的特性。

• single: 将整个集合作为一个分区。
• sample: 通过随机采样的方式来生成分区，快速但可能不均匀。
• split-vector: 通过 MongoDB 计算分片的 splitVector 命令来生成分区，快速且均匀。 仅适用于未分片集合，需要 splitVector 权限。
• sharded: 从 config.chunks 集合中直接读取分片集合的分片边界作为分区，不需要额外计算，快速且均匀。 仅适用于已经分片的集合，需要 config 数据库的读取权限。
• default: 对分片集合使用 sharded 策略，对未分片集合使用 split-vector 策略。

Lookup Cache

MongoDB 连接器可以用在时态表关联中作为一个可 lookup 的 source (又称为维表)，当前只支持同步的查找模式。

默认情况下，lookup cache 是未启用的，你可以将 lookup.cache 设置为 PARTIAL 参数来启用。

lookup cache 的主要目的是用于提高时态表关联 MongoDB 连接器的性能。默认情况下，lookup cache 不开启，所以所有请求都会发送到外部数据库。 当 lookup cache 被启用时，每个进程（即 TaskManager）将维护一个缓存。Flink 将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。 当缓存命中最大缓存行 lookup.partial-cache.max-rows 或当行超过 lookup.partial-cache.expire-after-write 或 lookup.partial-cache.expire-after-access 指定的最大存活时间时，缓存中的行将被设置为已过期。 缓存中的记录可能不是最新的，用户可以将缓存记录超时设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。

默认情况下，flink 会缓存主键的空查询结果，你可以通过将 lookup.partial-cache.caching-missing-key 设置为 false 来切换行为。

幂等写入

如果在 DDL 中定义了主键，MongoDB sink 将使用 upsert 语义而不是普通的 INSERT 语句。 我们将 DDL 中声明的主键进行组合作为 MongoDB 保留主键 _id，使用 upsert 模式进行写入，来保证写入的幂等性。

如果出现故障，Flink 作业会从上次成功的 checkpoint 恢复并重新处理，这可能导致在恢复过程中重复处理消息。 强烈推荐使用 upsert 模式，因为如果需要重复处理记录，它有助于避免违反数据库主键约束和产生重复数据。

过滤器下推

MongoDB 支持将 Flink SQL 的简单比较和逻辑过滤器下推以优化查询。 Flink SQL 过滤器到 MongoDB 查询操作符的映射如下表所示。

数据类型映射

MongoDB BSON 类型到 Flink SQL 数据类型的映射如下表所示。

对于 MongoDB 独有的类型，我们使用 Extended JSON 格式 将其映射为 Flink SQL STRING 类型。