# MongoDB Connecting Materialize to a MongoDB database for Change Data Capture (CDC). > For the complete documentation index, see [llms.txt](https://preview.materialize.com/materialize/40269/llms.txt). Using Change Data Capture (CDC), you can track and propagate changes from MongoDB to downstream consumers. This guide outlines how to ingest data from a MongoDB replica set (`rs0`) into Materialize; specifically from the collections `items` and `orders` in the `test` database. ## High-level architecture ```mermaid flowchart LR subgraph "Source" Mongo[("MongoDB (Replica Set/Sharded Cluster)")] end subgraph "Streaming Ingestion: Debezium + Kafka" Debezium["Debezium Connector"] Kafka{{"Kafka Broker"}} Schema["Schema Registry"] end subgraph "Materialize" MZ[("Materialize")] end Mongo -- "Change stream oplog data" --> Debezium Debezium -- "Writes Avro Events" --> Kafka Schema -. "Validates Schemas" .- Kafka Kafka -- "Consumes Topic" --> MZ Schema -. "Decodes Data" .- MZ ``` The architecture consists of the following components: - **Source Component: MongoDB** - **MongoDB**: MongoDB must be deployed as a **replica set** or **a sharded cluster** to generate an oplog (operations log), which records all data modifications. To deploy as a replica set or a sharded cluster, refer to the official MongoDB documentation on [replica sets](https://www.mongodb.com/docs/manual/replication/) and [sharded clusters](https://www.mongodb.com/docs/manual/sharding/). - **Streaming Ingestion Component: Debezium + Kafka (with Schema Registry)** - **Debezium:** A connector that [obtains the oplog data via MongoDB change streams](https://debezium.io/documentation//reference/stable/connectors/mongodb.html#change-streams). - **Kafka & Schema Registry:** The streaming transport layer. Debezium pushes the events to a Kafka topic, while the Schema Registry ensures the data structure (schema) is consistent and readable. - **Materialize:** Materialize connects to Kafka to ingest data from MongoDB. ## Prerequisites - **MongoDB:** Version 6.0 or later. Must be deployed as a [**replica set**](https://www.mongodb.com/docs/manual/replication/) or a [**sharded cluster**](https://www.mongodb.com/docs/manual/sharding/). - **Kafka Cluster:** A running Kafka broker and Schema Registry (e.g., Confluent Platform or Redpanda). - **Debezium Connect:** Version 2.5.4+ or later. A Kafka Connect cluster with the MongoDB connector plugin installed. ## A. Create a Debezium user in MongoDB. > **Important:** For Debezium to capture changes, your MongoDB deployment must be configured as a > replica set or a sharded cluster. Create a user (e.g., `debezium_materialize_user`) with the necessary permissions to read from the database. Depending on the connector's [`capture.scope`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#required-user-permissions) property, the user may need `read` on a specific database or all databases. For this guide, the example will ingest `items` and `orders` collections in the `test` database. As such, the example uses `capture.scope` of `database` to read from the `test` database only: ```javascript db.getSiblingDB("test").createUser({ user: "debezium_materialize_user", pwd: passwordPrompt(), // Or "" roles: [ { role: "read", db: "test" } ] }); ``` For simplicity, the user is created in the `test` database; however, the user can be created in a different database. ## B. Configure the streaming layer The streaming layer requires Kafka, Schema Registry, and Kafka Connect with the Debezium MongoDB connector. If you have not set up these services, see: - [Confluent Platform Quickstart](https://docs.confluent.io/platform/current/platform-quickstart.html)(*Recommended for Testing*): Docker-based setup for Kafka, Schema Registry, and Kafka Connect. - [Debezium Tutorial](https://debezium.io/documentation/reference/stable/tutorial.html): Example using Docker Compose. For this tutorial, ensure your Debezium container allows outbound traffic to your external MongoDB host. ### 1. Configure Kafka Connect worker Configure your Kafka Connect worker to use Avro serialization for Schema Registry integration; specifically, set the following environment variables/properties are set on your Connect worker: | Property | Value | |----------|-------| | `KEY_CONVERTER` | `io.confluent.connect.avro.AvroConverter` | | `VALUE_CONVERTER` | `io.confluent.connect.avro.AvroConverter` | | `CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL` | `http://schema-registry:8081` | | `CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL` | `http://schema-registry:8081` | ### 2. Configure the Debezium connector for MongoDB Register the MongoDB connector with the following configuration: | Parameter | Description | |-----------|-------------| | `name` | A unique name for the connector instance. | | [`connector.class`](https://docs.confluent.io/kafka-connectors/debezium-mongodb-source/current/mongodb_source_connector_config.html) | The Debezium MongoDB connector class, `"io.debezium.connector.mongodb.MongoDbConnector"`. | | [`mongodb.connection.string`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-mongodb-connection-string) | Your [MongoDB connection string](https://www.mongodb.com/docs/manual/reference/connection-string-formats/) with the `debezium_materialize_user`. | | [`topic.prefix`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-topic-prefix) | A unique prefix for Kafka topics. Topics are created as `..`. **Recommended** Once set, do not change the value of this property. See the [official documentation](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-topic-prefix) for details. | | [`collection.include.list`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-collection-include-list) | Comma-separated list of collections to capture in the format `.`, e.g., `"test.orders,test.items"`. | | [`capture.mode`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-capture-mode) | **Must** be `"change_streams_update_full"` (the default). This captures full document state on updates, which is required for Materialize's `UPSERT` envelope. | | [`capture.scope`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-capture-scope) | The scope of the change stream: `"database"` or `"deployment"`. This value affects the required MongoDB user permissions. | | [`capture.target`](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-capture-target) | The database to monitor for changes. Required only when `capture.scope` is `"database"`. | | [`transforms`](https://debezium.io/documentation/reference/stable/transformations/mongodb-event-flattening.html#mongodb-event-flattening-configuration) | Optional. Set to `"unwrap"` to extract the document state from Debezium's change event envelope. | | [`transforms.unwrap.type`](https://debezium.io/documentation/reference/stable/transformations/mongodb-event-flattening.html#mongodb-event-flattening-configuration) | Optional. Set to `"io.debezium.connector.mongodb.transforms.ExtractNewDocumentState"` to use the MongoDB-specific unwrap transform. | > **Important:** The `capture.mode` must be `change_streams_update_full`, the default. This > forces Debezium to send the entire document state for every change, which allows > Materialize to use the `UPSERT` envelope. 1. Create a `dbz_mongodb_connector.json` file with your connector configuration: ```json { "name": "mongodb-connector", "config": { "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", "mongodb.connection.string": "mongodb://debezium_materialize_user:@host1:27017,host2:27017,host3:27017/?replicaSet=rs0&authSource=test", "topic.prefix": "mdb-prod-rs0", "collection.include.list": "test.orders,test.items", "capture.mode": "change_streams_update_full", "capture.scope": "database", "capture.target": "test", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState" } } ``` **Required updates:** - Update `mongodb.connection.string` with your [MongoDB connection string](https://www.mongodb.com/docs/manual/reference/connection-string-formats/) for the `debezium_materialize_user`. For example: - `"mongodb+srv://debezium_materialize_user:@host[/[defaultauthdb][?options]]"` - `"mongodb://debezium_materialize_user:@host1[:port1][,...hostN[:portN]]/?replicaSet=&authSource=test"` - Update `collection.include.list` with the collections you want to capture. - Update `topic.prefix` with unique prefix for your environment. **Optional modifications:** - `capture.scope`: To monitor changes across all databases, you can change to `"deployment"` to monitor all databases (requires broader MongoDB permissions for your debezium user). - `capture.target`: Only needed if `capture.scope` is `"database"`. - `transforms` and `transforms.unwrap.type`: You can omit or customize. > **Tip:** When using the `unwrap` transform of type `ExtractNewDocumentState` with > MongoDB and Avro serialization, the Avro schema is inferred from the first > document processed. Because MongoDB allows the same field to have > different BSON types across documents, ensure that same field uses a > consistent BSON type across the collection to avoid schema inference > errors. > For example, if the first document has `price: 40`, the registered Avro > schema infers the `price` field type to be `int`. If a subsequent document > has `price: 2.25` (a decimal value), it will cause a schema mismatch error. > To avoid this, explicitly specify the BSON type in your documents using > `NumberDecimal()`, e.g., `price: NumberDecimal("40.00")` and `price: > NumberDecimal("2.25")`. > If you cannot enforce a consistent BSON type, you can omit the unwrap. 2. Register the connector with Kafka Connect: ```bash curl -X POST -H "Content-Type:application/json" \ http://:8083/connectors \ -d @dbz_mongodb_connector.json ``` Replace `` with your Kafka Connect hostname or IP address. ## C. Ingest data in Materialize ### 1. Create the Kafka and CSR connections. In Materialize, Create the [Kafka connection](https://preview.materialize.com/materialize/40269/markdown-docs/sql/create-connection/index.md#kafka) and the [Confluent Schema Registry connection](https://preview.materialize.com/materialize/40269/markdown-docs/sql/create-connection/index.md#confluent-schema-registry) 1. Create the [Kafka connection](https://preview.materialize.com/materialize/40269/markdown-docs/sql/create-connection/index.md#kafka). For example, if using `SASL_PLAINTEXT`: ```mzsql CREATE SECRET IF NOT EXISTS kafka_secret AS ''; CREATE CONNECTION IF NOT EXISTS kafka_connection TO KAFKA ( BROKER ':9092', SECURITY PROTOCOL = 'SASL_PLAINTEXT', SASL MECHANISMS = 'SCRAM-SHA-256', SASL USERNAME = '', SASL PASSWORD = SECRET kafka_secret ); ``` 1. Create the [Confluent Schema Registry connection](https://preview.materialize.com/materialize/40269/markdown-docs/sql/create-connection/index.md#confluent-schema-registry): ```mzsql CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY ( URL 'http://schema-registry:8081' ); ``` ### 2. Start ingesting data Create the sources for the specific Kafka topic (`..`). **Legacy Syntax:** With the legacy syntax, each source decodes its topic directly and is itself queryable. Picking up an upstream schema change requires dropping and recreating the source, which incurs downtime. ```mzsql CREATE SOURCE mdb_items FROM KAFKA CONNECTION kafka_connection (TOPIC 'mdb-prod-rs0.test.items') FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection ENVELOPE UPSERT; CREATE SOURCE mdb_orders FROM KAFKA CONNECTION kafka_connection (TOPIC 'mdb-prod-rs0.test.orders') FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection ENVELOPE UPSERT; ``` **New Syntax:** With the new syntax, create a source for each topic and then a table that decodes it. Each table pins its own reader schema, which lets you pick up upstream schema changes without downtime. For details, see [Handle upstream schema changes with zero downtime](https://preview.materialize.com/materialize/40269/markdown-docs/ingest-data/kafka/source-versioning/index.md). ```mzsql CREATE SOURCE mdb_items_src FROM KAFKA CONNECTION kafka_connection (TOPIC 'mdb-prod-rs0.test.items'); CREATE TABLE mdb_items FROM SOURCE mdb_items_src (REFERENCE "mdb-prod-rs0.test.items") FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection ENVELOPE UPSERT; CREATE SOURCE mdb_orders_src FROM KAFKA CONNECTION kafka_connection (TOPIC 'mdb-prod-rs0.test.orders'); CREATE TABLE mdb_orders FROM SOURCE mdb_orders_src (REFERENCE "mdb-prod-rs0.test.orders") FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection ENVELOPE UPSERT; ``` ### 3. Query the data Query the data using standard SQL. - To query the mdb_items: ```mzsql SELECT * FROM mdb_items; ``` > **Note:** The query includes the Kafka message `id` field as a column in the table. ```none id | _id | item_id | item | price | currency --------------------------+--------------------------+---------+-----------------------+-------+---------- 693c608a8de2a41c2ba365f3 | 693c608a8de2a41c2ba365f3 | 8 | donut | 1.25 | USD 693c608a8de2a41c2ba365ec | 693c608a8de2a41c2ba365ec | 1 | brownie | 2.25 | USD 693c608a8de2a41c2ba365f2 | 693c608a8de2a41c2ba365f2 | 7 | cupcake | 3.00 | USD 693c608a8de2a41c2ba365f4 | 693c608a8de2a41c2ba365f4 | 9 | egg tart | 2.50 | USD ``` > **Tip:** If you did not use the `unwrap` transform, the document is stored as a JSON > string in the `after` field in `mdb_items`. You can create a [parsing > view](https://preview.materialize.com/materialize/40269/markdown-docs/sql/types/jsonb/index.md#parsing) to map the individual document fields to columns instead. - To query the mdb_orders: ```mzsql SELECT * FROM mdb_orders; ``` > **Note:** The query includes the Kafka message `id` field as a column in the table. ```none id | _id | order_id | order_date | item | quantity | status --------------------------+--------------------------+----------+-------------------------+-----------------------+----------+------------ 693c608a8de2a41c2ba365d4 | 693c608a8de2a41c2ba365d4 | 3 | 2025-12-12 18:05:54.648 | donut | 36 | Pending 693c608a8de2a41c2ba365e3 | 693c608a8de2a41c2ba365e3 | 8 | 2025-12-10 18:50:54.648 | donut | 12 | Shipped 693c608a8de2a41c2ba365cc | 693c608a8de2a41c2ba365cc | 1 | 2025-12-12 18:35:54.648 | brownie | 10 | Pending 693c608a8de2a41c2ba365d2 | 693c608a8de2a41c2ba365d2 | 2 | 2025-12-12 18:20:54.648 | brownie | 20 | Pending ``` > **Tip:** If you did not use the `unwrap` transform, the document is stored as a JSON > string in the `after` field in `mdb_orders`. You can create a [parsing > view](https://preview.materialize.com/materialize/40269/markdown-docs/sql/types/jsonb/index.md#parsing) to map the individual document fields to > columns instead. ## Troubleshooting See also: [Debezium documentation on the MongoDB Connector](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-when-things-go-wrong)