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.

Prerequisites

Complete the 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.

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:

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:

{
  "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:

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:

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.

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:

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

[
  { _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:

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:

_id  amount
o1   19.99
o2   5.0
o3   42.5

Clean up

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

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