Skip to main content
Version: Next

MongoDB

MongoDB Sink Connector

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Key Features​

Tips

  1. If you want to use CDC-written features, enable the upsert-enable configuration.
  2. Enabling transaction is incompatible with the Zeta timer-flush feature. Pick one model per job; mixing them will silently disable the timer.

Description​

The MongoDB sink connector writes SeaTunnel rows into a MongoDB collection. Each row is converted to a BSON document and sent to the configured database and collection.

The connector supports two write semantics:

  • Append writes — every row produces a new document. Fast, but not idempotent across retries.
  • Upsert writes — when upsert-enable = true and primary-key is configured, the connector uses the primary key as the MongoDB _id (or compound _id) and upserts. Combined with checkpoint-based recovery this gives at-least-once with idempotent retries, which is the standard way to deliver exactly-once into MongoDB.

Buffering, retry, and the optional transaction are tuned via the options below.

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 SeaTunnel data type to MongoDB BSON type.

SeaTunnel Data TypeMongoDB BSON Type
STRINGObjectId
STRINGString
BOOLEANBoolean
BINARYBinary
INTEGERInt32
TINYINTInt32
SMALLINTInt32
BIGINTInt64
DOUBLEDouble
FLOATDouble
DECIMALDecimal128
DateDate
TimestampTimestamp[Date]
ROWObject
ARRAYArray

Tips

  1. When SeaTunnel writes Date and Timestamp types to MongoDB, both become MongoDB Date fields, but at different precisions: SeaTunnel Date is second precision; SeaTunnel Timestamp is millisecond precision.
  2. 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.

Sink Options​

NameTypeRequiredDefault ValueDescription
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 write to. When writing multiple tables from the source, you can use ${database_name} as a placeholder, for example database = "${database_name}_test_database".
collectionStringYes-The name of the MongoDB collection to write to. When writing multiple tables from the source, you can use ${database_name}, ${schema_name}, and ${table_name} as placeholders, for example collection = "${database_name}_${schema_name}_${table_name}_check".
buffer-flush.max-rowsIntNo1000The maximum number of buffered rows per batch request.
buffer-flush.intervalLongNo30000The maximum interval (in milliseconds) of buffered rows per batch request.
retry.maxIntNo3The maximum number of retries if writing records to MongoDB fails.
retry.intervalLongNo1000The retry interval (in milliseconds) if writing records to MongoDB fails.
upsert-enableBooleanNofalseWhether to write documents via upsert mode. When enabled, primary-key must also be configured.
primary-keyListNo-The primary keys used for upsert/update. The list format is ["id","name",...].
transactionBooleanNofalseWhether to use transactions in MongoSink (requires MongoDB 4.2+).
data_save_modeEnumNoAPPEND_DATAThe data saving mode for the MongoDB collection. Supported values: DROP_DATA (truncate the collection before writing), APPEND_DATA (append to existing data), ERROR_WHEN_DATA_EXISTS (fail if the collection already has data).
common-optionsNo-Sink plugin common parameters, please refer to Sink Common Options for details.

Tips​

  1. The connector-level data flushing logic is jointly controlled by three parameters: buffer-flush.max-rows, buffer-flush.interval, and checkpoint.interval. Whichever is reached first triggers the flush.
  2. The legacy option name upsert-key is still accepted as a fallback for primary-key. Do not set both at the same time.
  3. The transaction option is incompatible with the Zeta timer-flush feature described below. Enable exactly one model per job.

Zeta Timer Flush​

This engine-level feature is supported only by Zeta. Spark and Flink do not inject FlushSignal records. On Zeta, configure sink.flush.interval in the env block to flush pending bulk requests even when buffer-flush.max-rows has not been reached. Unlike buffer-flush.interval, the engine timer does not require a new input record to trigger the check.

Timer flush is enabled only when transaction = false. MongoDB transaction mode is committed through checkpoints, so timer flush is disabled to preserve the transaction boundary. The initial timer-flush implementation provides at-least-once delivery rather than 2PC exactly-once. Enabling upsert with deterministic primary keys can make retries idempotent.

env {
  job.mode = "STREAMING"
  checkpoint.interval = 300000
  sink.flush.interval = 5000
}

sink {
  MongoDB {
    uri = "mongodb://127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    buffer-flush.max-rows = 10000
    transaction = false
  }
}

How to Create a MongoDB Data Synchronization Job​

The following example writes randomly generated data into a MongoDB collection:

env {
  parallelism = 1
  job.mode = "BATCH"
  checkpoint.interval = 1000
}

source {
  FakeSource {
    row.num = 2
    bigint.min = 0
    bigint.max = 10000000
    split.num = 1
    split.read-interval = 300
    schema {
      fields {
        c_bigint = bigint
      }
    }
  }
}

sink {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test"
    collection = "test"
  }
}

Multiple Table Write​

When upstream records carry table metadata, database and collection can use placeholders. The common placeholders are ${database_name}, ${schema_name}, and ${table_name}.

source {
  FakeSource {
    tables_configs = [
      {
        schema = {
          table = "testDatabase1.testSchema1.testTable1"
          fields {
            id = int
            value = string
          }
        }
        rows = [
          {
            kind = INSERT
            fields = [1, "NEW"]
          }
        ]
      },
      {
        schema = {
          table = "testDatabase2.testSchema2.testTable2"
          fields {
            id = int
            amount = "decimal(16, 1)"
          }
        }
        rows = [
          {
            kind = INSERT
            fields = [1, 6.3]
          }
        ]
      }
    ]
  }
}

sink {
  MongoDB {
    uri = "mongodb://127.0.0.1:27017/test_db?retryWrites=true"
    database = "test_db"
    collection = "${database_name}_${schema_name}_${table_name}_check"
  }
}

Parameter Interpretation​

MongoDB Database Connection URI Examples​

Unauthenticated single node connection:

mongodb://127.0.0.1:27017/mydb

Replica set connection:

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

Authenticated replica set connection:

mongodb://admin:password@127.0.0.1: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.

Buffer Flush​

sink {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    buffer-flush.max-rows = 2000
    buffer-flush.interval = 1000
  }
}

Although MongoDB has fully supported multi-document transactions since version 4.2, this does not mean that every workload should use them. Transactions bring locking, node coordination, extra round trips, and performance overhead. The guiding principle is: avoid transactions whenever possible. A well-designed pipeline can usually make idempotent writes enough.

Idempotent Writes​

By specifying a clear primary key and using the upsert method, exactly-once write semantics can be achieved.

If primary-key and upsert-enable are defined in the configuration, the MongoDB sink uses upsert semantics instead of regular INSERT statements. The connector combines the primary keys declared in primary-key as the MongoDB reserved primary key and writes via upsert mode to ensure idempotent writes. In the event of a failure, SeaTunnel jobs recover from the last successful checkpoint and reprocess, which may result in duplicate processing during recovery. It is highly recommended to use upsert mode because it avoids violating database primary key constraints and generating duplicate data if records need to be reprocessed.

sink {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "users"
    upsert-enable = true
    primary-key = ["name", "status"]
  }
}

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 & Connector & 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