跳到主要内容
版本：3.0.0

MongoDB

MongoDB 源连接器

支持这些引擎​

Spark
Flink
SeaTunnel Zeta

关键特性​

描述​

MongoDB 源连接器从 MongoDB 集合中读取文档，并把每个 BSON 文档转换为 SeaTunnel 行记录。 它同时支持批和流两种作业模式，并通过 partition.split-key 把集合按取值范围切分为多个 split，实现并行读取。

在不扫描整张集合的前提下，可以缩小读取范围并控制返回字段：

  • 使用 match.query 过滤符合条件的文档。
  • 使用 match.projection 控制结果中返回的字段。
  • 使用 flat.sync-string 把整篇文档作为一条 JSON STRING 列读入，跳过固定 schema 的定义。

在流模式下，连接器读取已分配的 split，并通过 checkpoint 跟踪读取位置；任务重启后会从上次提交 的游标继续读取。

支持的数据源信息​

要使用 MongoDB 连接器，需要以下依赖。可以通过 install-plugin.sh 或 Maven 中央仓库下载。

数据源支持的版本依赖
MongoDB通用版本Download

数据类型映射​

下表列出了从 MongoDB BSON 类型到 SeaTunnel 数据类型的字段映射。

MongoDB BSON 类型SeaTunnel 数据类型
ObjectIdSTRING
StringSTRING
BooleanBOOLEAN
BinaryBINARY
Int32INTEGER
Int64BIGINT
DoubleDOUBLE
Decimal128DECIMAL
DateDate
TimestampTimestamp
ObjectROW
ArrayARRAY

针对 MongoDB 中的特殊类型，连接器使用扩展 JSON（Extended JSON）格式映射到 SeaTunnel 的 STRING 类型。

MongoDB BSON 类型SeaTunnel STRING
Symbol{"_value": {"$symbol": "12"}}
RegularExpression{"_value": {"$regularExpression": {"pattern": "^9$", "options": "i"}}}
JavaScript{"_value": {"$code": "function() { return 10; }"}}
DbPointer{"_value": {"$dbPointer": {"$ref": "db.coll", "$id": {"$oid": "63932a00da01604af329e33c"}}}}

提示

  1. 在 SeaTunnel 中使用 DECIMAL 类型时，最大精度不能超过 34 位。建议使用 decimal(34, 18) 以满足支持的精度与标度。

源配置项​

参数名称类型是否必填默认值描述
uriString是-MongoDB 标准连接 URI，例如 mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true。更多示例请参考 参数说明。
databaseString是-要读取的 MongoDB 数据库名称。
collectionString是-要读取的 MongoDB 集合名称。
schemaConfig是-MongoDB 的 BSON 与 SeaTunnel 数据结构的映射。更多详情请参考 Schema 特性。
match.queryString否-用于过滤读取文档的 MongoDB 查询表达式。兼容旧版参数名 matchQuery。
match.projectionString否-用于控制查询结果中包含字段的 MongoDB 投影表达式。
partition.split-keyString否_id用作 MongoDB 分片字段的列名，连接器会按该字段的取值范围切分集合。
partition.split-sizeLong否64 1024 1024每个 MongoDB split 的大小。split 越小并行度越高，split 越大并行度越低。
cursor.no-timeoutBoolean否trueMongoDB 服务端默认会在游标空闲 10 分钟后关闭游标以回收内存。将此选项设置为 true 可让游标在长时间运行的批次中保持打开。如果应用持有批次超过 30 分钟，MongoDB 会将当前会话标记为过期并关闭。
fetch.sizeInt否2048每批从服务器获取的文档数。合理设置可以提升查询性能并降低一次性获取大量数据带来的内存压力。
max.time-minLong否10每次 MongoDB 查询的最大执行时间（分钟）。超过该限制 MongoDB 将终止操作并返回错误。
flat.sync-stringBoolean否false开启后，连接器会把整篇 MongoDB 文档映射到一个 SeaTunnel STRING 字段。此时 schema 只能声明一个字段，且该字段必须是 STRING 类型。
common-options否-源插件通用参数，详见 源通用选项。

提示​

  1. match.query 与旧版参数名 matchQuery 等价，二者不能同时设置。
  2. 使用 partition.split-key 时建议选择有索引的字段，能显著加快 split 边界扫描。
  3. 当 flat.sync-string = true 时，schema 仅用于声明单个接收文档的 STRING 字段。

如何创建 MongoDB 数据同步作业​

下面的示例从 MongoDB 读取数据并打印到本地客户端：

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "source_table"
    schema = {
      fields {
        c_map = "map<string, string>"
        c_array = "array<int>"
        c_string = string
        c_boolean = boolean
        c_int = int
        c_bigint = bigint
        c_double = double
        c_bytes = bytes
        c_date = date
        c_decimal = "decimal(34, 18)"
        c_timestamp = timestamp
        c_row = {
          c_map = "map<string, string>"
          c_array = "array<int>"
          c_string = string
          c_boolean = boolean
          c_int = int
          c_bigint = bigint
          c_double = double
          c_bytes = bytes
          c_date = date
          c_decimal = "decimal(34, 18)"
          c_timestamp = timestamp
        }
      }
    }
  }
}

sink {
  Console {
    parallelism = 1
  }
}

参数说明​

MongoDB 数据库连接 URI 示例​

无认证的单节点连接：

mongodb://192.168.0.100:27017/mydb

副本集连接：

mongodb://192.168.0.100:27017/mydb?replicaSet=xxx

带认证的副本集连接：

mongodb://admin:password@192.168.0.100:27017/mydb?replicaSet=xxx&authSource=admin

多节点副本集连接：

mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb?replicaSet=xxx

分片集群连接（通过一个 mongos 路由）：

mongodb://mongos1.example.com:27017,mongos2.example.com:27017,mongos3.example.com:27017/mydb

多个 mongos 节点连接：

mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb

注意：URI 中的用户名与密码在拼接到连接字符串前必须进行 URL 编码。

匹配查询扫描​

在数据同步场景中，建议尽早使用 match.query 减少下游算子需要处理的文档数量，从而提升整体 性能。下面是一个简单的示例：

source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "orders"
    match.query = "{status: \"A\"}"
    schema = {
      fields {
        id = bigint
        status = string
      }
    }
  }
}

下面是常见数据类型对应的 match.query 表达式：

# 布尔类型
"{c_boolean: true}"
# 字符串类型
"{c_string: \"OCzCj\"}"
# 整数类型
"{c_int: 2}"
# 日期类型
"{c_date: {\$date: \"2023-06-26T16:00:00.000Z\"}}"
# 浮点类型
"{c_double: {\$gte: 1.71763202185342e+308}}"

完整查询语法请参考 MongoDB 官方文档： https://www.mongodb.com/docs/manual/tutorial/query-documents

投影扫描​

MongoDB 中的 Projection 用来控制查询结果中返回哪些字段：在 find() 方法中通过第二个参数 传入一个投影对象，键表示字段，值 1 表示包含，0 表示排除。例如对于 users 集合：

// 仅返回 `name` 字段，过滤掉 `email` 字段
db.users.find({}, { name: 1, email: 0 });

在数据同步场景中，尽早使用 Projection 可以减少下游算子需要处理的字段数量。下面是 SeaTunnel 中使用投影的简单示例：

source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    match.projection = "{ name: 1, email: 0 }"
    schema = {
      fields {
        name = string
      }
    }
  }
}

分区扫描​

为了加速并行源任务中的数据读取，SeaTunnel 为 MongoDB 集合提供了分区扫描能力。通过 partition.split-key 指定分片字段、partition.split-size 指定每个 split 的大小，可以控制 数据分片方式：

source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    partition.split-key = "id"
    partition.split-size = 1024
    schema = {
      fields {
        id = bigint
        status = string
      }
    }
  }
}

建议选择有索引的字段作为 split key，能显著加快 split 边界扫描。

Flat Sync String​

启用 flat.sync-string 后，只需声明一个 STRING 类型字段，连接器会把每条 MongoDB 文档序列化 为扩展 JSON 字符串写入该字段。

env {
  parallelism = 1
  job.mode = "BATCH"
}
source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    flat.sync-string = true
    schema = {
      fields {
        data = string
      }
    }
  }
}
sink {
  Console {}
}

通过该配置写入的样本数据示例：

{
  "_id": {
    "$oid": "643d41f5fdc6a52e90e59cbf"
  },
  "c_map": {
    "OQBqH": "jllt",
    "rkvlO": "pbfdf",
    "pCMEX": "hczrdtve",
    "DAgdj": "t",
    "dsJag": "voo"
  },
  "c_array": [
    { "$numberInt": "-865590937" },
    { "$numberInt": "833905600" },
    { "$numberInt": "-1104586446" },
    { "$numberInt": "2076336780" },
    { "$numberInt": "-1028686444" }
  ],
  "c_string": "bddkzxr",
  "c_boolean": false,
  "c_tinyint": { "$numberInt": "39" },
  "c_smallint": { "$numberInt": "23672" },
  "c_int": { "$numberInt": "-495763561" },
  "c_bigint": { "$numberLong": "3768307617923954543" },
  "c_double": { "$numberDouble": "1.1706091642478246E308" },
  "c_bytes": { "$binary": { "base64": "ZWJ4", "subType": "00" } },
  "c_date": { "$date": { "$numberLong": "1686614400000" } },
  "c_decimal": { "$numberDecimal": "683265300" },
  "c_timestamp": { "$date": { "$numberLong": "1684283772000" } }
}

修改日志​

Change Log
ChangeCommitVersion
[Improve][Common] Add HashUtils.bucketIndex for hash-to-bucket routing (#11987)https://github.com/apache/seatunnel/commit/be53a1d3d3.0.0
[Improve][Connector-V2][MongoDB] Migrate fetch size validation to declarative OptionRule (#11886)https://github.com/apache/seatunnel/commit/f4840f1c73.0.0
[Fix][Connector-V2] Fix MongoDB reader cursor cleanup (#11887)https://github.com/apache/seatunnel/commit/7d5ee2e883.0.0
[Feature][Connector-V2] Add MongoDB timer flush (#11527)https://github.com/apache/seatunnel/commit/15ebb211e3.0.0
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[fix][connector-mango] fix split with avgSize zero error (#9255)https://github.com/apache/seatunnel/commit/564863b9332.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Fix][MongoDB] The Long type cannot handle string values in scientific notation (#8783)https://github.com/apache/seatunnel/commit/00f550e3d02.3.11
[Improve] sink mongodb schema is not required (#8887)https://github.com/apache/seatunnel/commit/3cfe8c12b92.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Fix][Connector-Mongodb] close MongodbClient when close MongodbReader (#8592)https://github.com/apache/seatunnel/commit/06b2fc0e062.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Bug][connectors-v2] fix mongodb bson convert exception (#8044)https://github.com/apache/seatunnel/commit/b222c13f2f2.3.9
[Hotfix][Connector-v2] Fix the ClassCastException for connector-mongodb (#7586)https://github.com/apache/seatunnel/commit/dc43370e8c2.3.8
[Improve][Test][Connector-V2][MongoDB] Add few test cases for BsonToRowDataConverters (#7579)https://github.com/apache/seatunnel/commit/a797041e5d2.3.8
[Improve][Connector-V2][MongoDB] A BsonInt32 will be convert to a long type (#7567)https://github.com/apache/seatunnel/commit/adf26c20c52.3.8
[Improve][Connector-V2][MongoDB] Support to convert to double from any numeric type (#6997)https://github.com/apache/seatunnel/commit/c5159a27602.3.6
[bugfix][connector-mongodb] fix mongodb null value write (#6967)https://github.com/apache/seatunnel/commit/c5ecda50f82.3.6
[Improve][MongoDB] Implement TableSourceFactory to create mongodb source (#5813)https://github.com/apache/seatunnel/commit/59cccb60972.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[bugfix][mongodb] Fixed unsupported exception caused by bsonNull (#5659)https://github.com/apache/seatunnel/commit/cab864aa4d2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)https://github.com/apache/seatunnel/commit/ed5eadcf732.3.3
[Improve][Connector-v2][Mongodb]sink support transaction update/writing (#5034)https://github.com/apache/seatunnel/commit/b1203c905e2.3.3
[Hotfix][Connector-V2][Mongodb] Compatible with historical parameters (#4997)https://github.com/apache/seatunnel/commit/31db35bee72.3.3
[Improve][Connector-v2][Mongodb]Optimize reading logic (#5001)https://github.com/apache/seatunnel/commit/830196d8b72.3.3
[Hotfix][Connector-V2][Mongodb] Fix document error content and remove redundant code (#4982)https://github.com/apache/seatunnel/commit/526197af672.3.3
[Feature][connector-v2][mongodb] mongodb support cdc sink (#4833)https://github.com/apache/seatunnel/commit/cb651cd7f32.3.3
[Feature][Connector-v2][Mongodb]Refactor mongodb connector (#4620)https://github.com/apache/seatunnel/commit/5b1a843e402.3.2
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API &amp; Connector &amp; Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Improve] mongodb connector v2 add source query capability (#3697)https://github.com/apache/seatunnel/commit/8a7fe6fcb62.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][MongoDB] Unified exception for MongoDB source & sink connector (#3522)https://github.com/apache/seatunnel/commit/5af632e32b2.3.0
[Feature][Connector V2] expose configurable options in MongoDB (#3347)https://github.com/apache/seatunnel/commit/ffd5778efc2.3.0
[Improve][all] change Log to @Slf4j (#3001)https://github.com/apache/seatunnel/commit/6016100f122.3.0-beta
[Improve][Connector-V2] Improve mongodb connector (#2778)https://github.com/apache/seatunnel/commit/efbf793fa52.2.0-beta
[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)https://github.com/apache/seatunnel/commit/cbf82f755c2.2.0-beta
[Feature][Connector-V2] Add mongodb connecter sink (#2694)https://github.com/apache/seatunnel/commit/51c28a33872.2.0-beta
[Feature][Connector-V2] Add mongodb connecter source (#2596)https://github.com/apache/seatunnel/commit/3ee8a8a6192.2.0-beta