<a id="cmf-sql-example-mongodb"></a>

# Read and Write MongoDB

This example uses the off-the-shelf `flink-sql-connector-mongodb` connector for
Flink to write rows to a MongoDB collection and read them back. The connector is a
released JAR from Maven Central and needs no custom code, so it is a good model
for adding any third-party Flink SQL connector. For the artifact mechanism, see
[Custom Connectors and Formats in Confluent Manager for Apache Flink](../custom-connectors.md#cmf-sql-custom-connectors).

## Prerequisites

Complete the [examples setup](setup.md#cmf-sql-examples-setup) first. It provides the
`test` environment, the `pool` compute pool, and artifact management. This
example also needs:

- A MongoDB instance reachable from the Flink cluster (installed below).
- The `flink-sql-connector-mongodb` JAR that matches the Flink cluster version,
  for example `flink-sql-connector-mongodb-2.0.0-1.19.jar` from Maven Central.

## Install MongoDB

For a self-contained demo, run a single `mongo:7` instance in the cluster. This
manifest, `mongodb.yaml`, creates the deployment and a `mongo-service` the Flink
cluster reaches by its in-cluster DNS name. It has no persistence, so it is for
demos only. Substitute your own MongoDB for anything you want to keep.

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mongo-test
  namespace: flink
spec:
  replicas: 1
  selector:
    matchLabels:
      app: mongo-test
  template:
    metadata:
      labels:
        app: mongo-test
    spec:
      containers:
        - name: mongo
          image: mongo:7
          ports:
            - containerPort: 27017
          env:
            - name: MONGO_INITDB_ROOT_USERNAME
              value: admin
            - name: MONGO_INITDB_ROOT_PASSWORD
              value: mongo-pass
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              memory: 512Mi
---
apiVersion: v1
kind: Service
metadata:
  name: mongo-service
  namespace: flink
spec:
  selector:
    app: mongo-test
  ports:
    - port: 27017
      targetPort: 27017
```

Apply it and wait for the pod to be ready:

```bash
kubectl apply -f mongodb.yaml
kubectl -n flink rollout status deploy/mongo-test
```

MongoDB is then reachable in-cluster at
`mongo-service.flink.svc.cluster.local:27017` with the `admin` / `mongo-pass`
credentials from the manifest.

## Upload the connector artifact

Label the JAR as a `connector` with the `mongodb` identifier, in
`artifact.json`:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "mongodb-connector.jar",
    "labels": {
      "cmf.confluent.io/sql-artifact-type": "connector",
      "cmf.confluent.io/sql-artifact-identifier": "mongodb"
    }
  },
  "spec": {}
}
```

Upload it:

```bash
curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@flink-sql-connector-mongodb-2.0.0-1.19.jar'
```

## Create the table

Create a table backed by a MongoDB collection. Point the `uri` at the service
from the manifest. The connector requires a primary key for its upsert semantics,
so declare one:

```bash
confluent flink statement create mongo-create \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool pool \
  --catalog _env_test \
  --database default \
  --sql "CREATE TABLE \`mongo_orders\` (\`_id\` STRING, \`amount\` DOUBLE, PRIMARY KEY (\`_id\`) NOT ENFORCED) WITH ('connector' = 'mongodb', 'uri' = 'mongodb://admin:mongo-pass@mongo-service.flink.svc.cluster.local:27017/?authSource=admin', 'database' = 'testdb', 'collection' = 'orders');"
```

## Write rows

Insert three rows. On first use of the table, CMF resolves the `mongodb`
artifact and ships it to the Flink cluster. The bounded `INSERT` reaches
`COMPLETED` when the documents are written.

```bash
confluent flink statement create mongo-insert \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool pool \
  --catalog _env_test \
  --database default \
  --sql "INSERT INTO \`mongo_orders\` VALUES ('o1', 19.99), ('o2', 5.00), ('o3', 42.50);"
```

Confirm the documents directly in MongoDB with `mongosh` in the `mongo-test`
pod:

```bash
kubectl -n flink exec deploy/mongo-test -- \
  mongosh "mongodb://admin:mongo-pass@localhost:27017/testdb?authSource=admin" \
  --quiet --eval "db.orders.find().toArray()"
```

```text
[
  { _id: 'o1', amount: 19.99 },
  { _id: 'o2', amount: 5 },
  { _id: 'o3', amount: 42.5 }
]
```

## Read rows back

Query the same table to read the documents through Flink:

```bash
confluent flink statement create mongo-select \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool pool \
  --catalog _env_test \
  --database default \
  --sql "SELECT * FROM \`mongo_orders\`;"
```

The query returns the three rows:

```text
_id  amount
o1   19.99
o2   5.0
o3   42.5
```

## Clean up

Drop the table, then remove the connector and the MongoDB deployment:

```bash
confluent flink statement create mongo-drop \
  --url http://cmf:8080 --environment test --compute-pool pool \
  --catalog _env_test --database default \
  --sql "DROP TABLE IF EXISTS \`mongo_orders\`;"

curl -X DELETE http://cmf:8080/cmf/api/v1/environments/test/artifacts/mongodb-connector.jar
kubectl delete -f mongodb.yaml
```

## Related content

- [Custom Connectors and Formats in Confluent Manager for Apache Flink](../custom-connectors.md#cmf-sql-custom-connectors)
- [Round-Trip an Apache Iceberg Table](iceberg.md#cmf-sql-example-iceberg)
- [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts)
- [Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples)