MongoDB

1db.runCommand({
2  collMod: "products",
3  changeStreamPreAndPostImages: { enabled: true }
4})

1db.createCollection("products", {
2  changeStreamPreAndPostImages: { enabled: true }
3})

1CREATE TEMPORARY TABLE mongo_full_cdc (
2  `_id`   STRING,
3  name    STRING,
4  weight  DECIMAL,
5  PRIMARY KEY (`_id`) NOT ENFORCED
6) WITH (
7  'connector'           = 'mongodb-cdc',
8  'hosts'               = '<host>:27017',
9  'username'            = '<username>',
10  'password'            = '${secret_values.password}',
11  'database'            = 'myDatabase',
12  'collection'          = 'products',
13  'scan.full-changelog' = 'true'
14);

1CREATE TABLE tableName (
2  _id STRING,
3  -- additional columns
4  PRIMARY KEY (_id) NOT ENFORCED
5) WITH (
6  'connector'  = 'mongodb',
7  'hosts'      = '<host>:27017',
8  'username'   = '<username>',
9  'password'   = '${secret_values.password}',
10  'database'   = '<database>',
11  'collection' = '<collection>'
12);

1CREATE TEMPORARY TABLE mongo_source (
2  `_id`   STRING,
3  name    STRING,
4  weight  DECIMAL,
5  PRIMARY KEY (`_id`) NOT ENFORCED
6) WITH (
7  'connector'  = 'mongodb',
8  'hosts'      = '<host>:27017',
9  'username'   = '<username>',
10  'password'   = '${secret_values.password}',
11  'database'   = 'myDatabase',
12  'collection' = 'products'
13);
14
15SELECT * FROM mongo_source;

1CREATE TEMPORARY TABLE mongo_cdc_source (
2  `_id`   STRING,
3  name    STRING,
4  weight  DECIMAL,
5  PRIMARY KEY (`_id`) NOT ENFORCED
6) WITH (
7  'connector'  = 'mongodb-cdc',
8  'hosts'      = '<host>:27017',
9  'username'   = '<username>',
10  'password'   = '${secret_values.password}',
11  'database'   = 'myDatabase',
12  'collection' = 'products'
13);
14
15CREATE TEMPORARY TABLE products_sink (
16  `_id`   STRING,
17  name    STRING,
18  weight  DECIMAL
19) WITH (
20  'connector' = 'print',
21  'logger'    = 'true'
22);
23
24INSERT INTO products_sink
25SELECT _id, name, weight
26FROM mongo_cdc_source;

1CREATE TEMPORARY TABLE datagen_source (
2  `_id`   STRING,
3  name    STRING,
4  weight  DECIMAL
5) WITH (
6  'connector' = 'datagen'
7);
8
9CREATE TEMPORARY TABLE mongo_sink (
10  `_id`   STRING,
11  name    STRING,
12  weight  DECIMAL,
13  PRIMARY KEY (`_id`) NOT ENFORCED
14) WITH (
15  'connector'                  = 'mongodb',
16  'hosts'                      = '<host>:27017',
17  'username'                   = '<username>',
18  'password'                   = '${secret_values.password}',
19  'database'                   = 'myDatabase',
20  'collection'                 = 'products',
21  'sink.buffer-flush.max-rows' = '500',
22  'sink.buffer-flush.interval' = '2s',
23  'sink.delivery-guarantee'    = 'at-least-once',
24  'sink.max-retries'           = '5'
25);
26
27INSERT INTO mongo_sink SELECT * FROM datagen_source;

1CREATE TEMPORARY TABLE datagen_source (
2  id         STRING,
3  a          INT,
4  b          BIGINT,
5  `proctime` AS PROCTIME()
6) WITH (
7  'connector' = 'datagen'
8);
9
10CREATE TEMPORARY TABLE mongo_dim (
11  `_id`   STRING,
12  name    STRING,
13  weight  DECIMAL,
14  PRIMARY KEY (`_id`) NOT ENFORCED
15) WITH (
16  'connector'  = 'mongodb',
17  'hosts'      = '<host>:27017',
18  'username'   = '<username>',
19  'password'   = '${secret_values.password}',
20  'database'   = 'myDatabase',
21  'collection' = 'products'
22);
23
24CREATE TEMPORARY TABLE enriched_sink (
25  id     STRING,
26  a      INT,
27  b      BIGINT,
28  name   STRING
29) WITH (
30  'connector' = 'print',
31  'logger'    = 'true'
32);
33
34INSERT INTO enriched_sink
35SELECT T.id, T.a, T.b, H.name
36FROM datagen_source AS T
37JOIN mongo_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H
38ON T.id = H._id;

1CREATE TEMPORARY TABLE mongo_full_cdc (
2  `_id`   STRING,
3  name    STRING,
4  weight  DECIMAL,
5  PRIMARY KEY (`_id`) NOT ENFORCED
6) WITH (
7  'connector'           = 'mongodb-cdc',
8  'hosts'               = '<host>:27017',
9  'username'            = '<username>',
10  'password'            = '${secret_values.password}',
11  'database'            = 'myDatabase',
12  'collection'          = 'products',
13  'scan.full-changelog' = 'true'
14);
15
16CREATE TEMPORARY TABLE products_sink (
17  `_id`   STRING,
18  name    STRING,
19  weight  DECIMAL
20) WITH (
21  'connector' = 'print',
22  'logger'    = 'true'
23);
24
25INSERT INTO products_sink
26SELECT _id, name, weight
27FROM mongo_full_cdc;

1MongoDBSource.builder()
2  .hosts("<host>:27017")
3  .username("<username>")
4  .password("<password>")
5  .databaseList("myDatabase")
6  .collectionList("products")
7  .startupOptions(StartupOptions.initial())
8  .deserializer(new JsonDebeziumDeserializationSchema())
9  .build();

1MongoSource.<RowData>builder()
2  .setUri("mongodb://<username>:<password>@<host>:27017")
3  .setDatabase("myDatabase")
4  .setCollection("products")
5  .setDeserializationSchema(new MongoDeserializationSchema<RowData>() {
6      @Override
7      public RowData deserialize(BsonDocument document) {
8          // Convert BsonDocument to RowData
9      }
10      @Override
11      public TypeInformation<RowData> getProducedType() {
12          return Types.ROW_NAMED(...);
13      }
14  })
15  .build();

1MongoSink.<RowData>builder()
2  .setUri("mongodb://<username>:<password>@<host>:27017")
3  .setDatabase("myDatabase")
4  .setCollection("products")
5  .setBatchSize(1000)
6  .setBatchIntervalMs(1000)
7  .setMaxRetries(3)
8  .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
9  .setSerializationSchema(
10      new MongoSerializationSchema<RowData>() {
11          @Override
12          public BsonDocument serialize(RowData element) {
13              // Convert RowData to BsonDocument
14          }
15      })
16  .build();