Skip to main content
Version: Next

MongoDB

MongoDB Source Connector

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Key Features​

Description​

The MongoDB source connector reads documents from a MongoDB collection and converts each BSON document into a SeaTunnel row. It supports both batch and streaming jobs, and reads in parallel by splitting the source collection across partition.split-key ranges.

You can narrow what is read and what columns are returned without scanning the whole collection:

  • Use match.query to filter documents by a MongoDB query expression.
  • Use match.projection to control which fields appear in the result.
  • Use flat.sync-string to capture the whole document as one JSON STRING column when no fixed schema is needed.

In streaming mode the connector reads the assigned splits and tracks progress through checkpoints so a restarted job resumes from the last committed cursor.

Supported DataSource Info​

In order to use the MongoDB connector, the following dependency is required. It can be downloaded via install-plugin.sh or from the Maven central repository.

DatasourceSupported VersionsDependency
MongoDBUniversalDownload

Data Type Mapping​

The following table lists the field data type mapping from MongoDB BSON type to SeaTunnel data type.

MongoDB BSON typeSeaTunnel Data Type
ObjectIdSTRING
StringSTRING
BooleanBOOLEAN
BinaryBINARY
Int32INTEGER
Int64BIGINT
DoubleDOUBLE
Decimal128DECIMAL
DateDate
TimestampTimestamp
ObjectROW
ArrayARRAY

For specific types in MongoDB, the connector uses Extended JSON format and maps them to the SeaTunnel STRING type.

MongoDB BSON typeSeaTunnel 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"}}}}

Tips

  1. When using the DECIMAL type in SeaTunnel, the maximum range cannot exceed 34 digits. Use decimal(34, 18) to stay within the supported precision and scale.

Source Options​

NameTypeRequiredDefaultDescription
uriStringYes-The MongoDB standard connection URI, for example mongodb://user:password@hosts:27017/database?readPreference=secondary&slaveOk=true. See Parameter Interpretation for more URI samples.
databaseStringYes-The name of the MongoDB database to read from.
collectionStringYes-The name of the MongoDB collection to read from.
schemaConfigYes-The mapping between MongoDB's BSON and SeaTunnel data structure. For more details, see Schema Feature.
match.queryStringNo-MongoDB query expression used to filter documents for read operations. Compatible with the legacy option name matchQuery.
match.projectionStringNo-MongoDB projection expression used to control which fields appear in the result.
partition.split-keyStringNo_idThe field used as the MongoDB split key. The connector splits the collection by the value range of this key.
partition.split-sizeLongNo64 1024 1024The size of each MongoDB split. Smaller split sizes increase the number of splits and parallelism, while larger sizes reduce it.
cursor.no-timeoutBooleanNotrueMongoDB server normally times out idle cursors after 10 minutes of inactivity to reclaim memory. Set this option to true to keep the cursor open across long-running batches. If the application holds the batch for more than 30 minutes, MongoDB marks the session as expired and closes it.
fetch.sizeIntNo2048The number of documents obtained from the server per batch. Tuning this value balances query performance and memory pressure.
max.time-minLongNo10The maximum execution time (in minutes) for each MongoDB query. MongoDB terminates the operation and returns an error when this limit is exceeded.
flat.sync-stringBooleanNofalseWhen enabled, the connector maps the whole MongoDB document into one SeaTunnel STRING field. The schema must contain exactly one field and that field must be of type STRING.
common-optionsNo-Source plugin common parameters, please refer to Source Common Options for details.

Tips​

  1. The match.query option is compatible with the legacy option name matchQuery; they are equivalent.
  2. Use partition.split-key together with partition.split-size to control parallel reads. The split key should reference an indexed field for best performance.
  3. When flat.sync-string = true, the configured schema is ignored except for the single STRING field that receives the document.

How to Create a MongoDB Data Synchronization Job​

The following example reads data from MongoDB and prints it on the local client:

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
  }
}

Parameter Interpretation​

MongoDB Database Connection URI Examples​

Unauthenticated single node connection:

mongodb://192.168.0.100:27017/mydb

Replica set connection:

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

Authenticated replica set connection:

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

Multi-node replica set connection:

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

Sharded cluster connection (route through one mongos):

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

Multiple mongos connections (comma-separated list of mongos hosts):

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

Note: The username and password in the URI must be URL-encoded before being concatenated into the connection string.

MatchQuery Scan​

In data synchronization scenarios, the match.query approach should be used early to reduce the number of documents that need to be processed by downstream operators, improving overall performance. Here is a simple example of using 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
      }
    }
  }
}

The following are examples of match.query expressions for various data types:

# Query Boolean type
"{c_boolean: true}"
# Query string type
"{c_string: \"OCzCj\"}"
# Query the integer
"{c_int: 2}"
# Query the date type
"{c_date: {\$date: \"2023-06-26T16:00:00.000Z\"}}"
# Query the floating point type
"{c_double: {\$gte: 1.71763202185342e+308}}"

Refer to the MongoDB manual for the full query syntax: https://www.mongodb.com/docs/manual/tutorial/query-documents

Projection Scan​

In MongoDB, projection controls which fields appear in the query results by specifying which fields are returned and which are excluded. In the find() method, a projection object can be passed as the second argument. A value of 1 includes the field, 0 excludes it. For example, given a users collection:

// Returns only the `name` field and excludes the `email` field
db.users.find({}, { name: 1, email: 0 });

In data synchronization scenarios, projection should be used early to reduce the number of fields that need to be processed by downstream operators. Here is a simple example of using projection in 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
      }
    }
  }
}

Partitioned Scan​

To speed up reading data in parallel source tasks, SeaTunnel provides a partitioned scan feature for MongoDB collections. Configure partition.split-key for the split field and partition.split-size for the split size to control data sharding:

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
      }
    }
  }
}

Tip: Pick a split key backed by an index so the source can fast-skip through the collection when enumerating split ranges.

Flat Sync String​

By enabling flat.sync-string, you only need to declare a single field whose type is STRING. The connector will serialize each MongoDB document as an Extended JSON string and put it into that field.

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 {}
}

A sample document that flows through this configuration looks like the following:

{
  "_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" } }
}

Changelog​

Change Log
ChangeCommitVersion
[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