MongoDB CDC

Synchronize one MongoDB collection into a Paimon table, or selected collections into a Paimon database.

See CDC Action Configuration for command syntax, computed columns, and job settings, and Schema Evolution and Type Mapping for shared behavior.

Prerequisites

Place these dependencies in <FLINK_HOME>/lib/ alongside the Paimon Flink bundled jar. The dependency baseline for this Paimon version is Flink CDC 3.5.0.

flink-sql-connector-mongodb-cdc-3.5.0.jar

Synchronizing Tables

By using MongoDBSyncTableAction in a Flink DataStream job or directly through flink run, users can synchronize one collection from MongoDB into one Paimon table.

Schema and Document Identity

Source fields are mapped to STRING; computed columns use the result type of their function. The action uses _id to identify records. Updates carry the resulting document, while delete events may contain only the document identity, so other document fields cannot replace _id as the record key.

`--mongodb_conf` option	Behavior
`schema.start.mode=dynamic` (default)	Infer the initial fields from the first document in the collection. The collection must contain a document when starting the table action. Add new top-level fields as they appear; nested values remain within their source fields.
`schema.start.mode=specified`	Select fields with `field.name` and corresponding JSONPath expressions with `parser.path`. Include `_id` and provide one path per field, in the same order.
`default.id.generation=true`	Unwrap the `$oid` representation of `_id`. Set to `false` to retain its original structure.

If the Paimon table you specify does not exist, this action will automatically create the table. Its schema will be derived from MongoDB collection.

Example 1: synchronize collection into one Paimon table

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_table \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --mongodb_conf hosts=127.0.0.1:27017 \
    --mongodb_conf username=root \
    --mongodb_conf password=123456 \
    --mongodb_conf database=source_db \
    --mongodb_conf collection=source_table1 \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4

Example 2: Synchronize collection into a Paimon table according to the specified field mapping.

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_table \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --mongodb_conf hosts=127.0.0.1:27017 \
    --mongodb_conf username=root \
    --mongodb_conf password=123456 \
    --mongodb_conf database=source_db \
    --mongodb_conf collection=source_table1 \
    --mongodb_conf schema.start.mode=specified \
    --mongodb_conf field.name=_id,name,description \
    --mongodb_conf 'parser.path=$._id,$.name,$.description' \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4

Table Action Reference

Command syntax (square brackets indicate optional arguments):

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_table \
    --warehouse <warehouse-path> \
    --database <database-name> \
    --table <table-name> \
    [--partition_keys <partition_keys>] \
    [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
    [--mongodb_conf <mongodb-cdc-source-conf> [--mongodb_conf <mongodb-cdc-source-conf> ...]] \
    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
    [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]

Configuration	Description
--warehouse	The path to Paimon warehouse.
--database	The database name in Paimon catalog.
--table	The Paimon table name.
--partition_keys	The partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example "dt,hh,mm".
--computed_column	The definitions of computed columns. The argument field is from MongoDB collection field name. See here for a complete list of configurations.
--mongodb_conf	The configuration for Flink CDC MongoDB sources. Each configuration should be specified in the format "key=value". hosts, username, password, database and collection are required configurations, others are optional. See its document for a complete list of configurations.
--catalog_conf	The configuration for Paimon catalog. Each configuration should be specified in the format "key=value". See here for a complete list of catalog configurations.
--table_conf	The configuration for Paimon table sink. Each configuration should be specified in the format "key=value". See here for a complete list of table configurations.

Synchronizing Databases

By using MongoDBSyncDatabaseAction in a Flink DataStream job or directly through flink run, users can synchronize the whole MongoDB database into one Paimon database.

Each collection maps to its own target table, with _id as the source primary key. The action creates missing tables from the corresponding collection's events and evolves existing schemas. New collections that match the configured filters can be included while the job runs.

Example 1: synchronize entire database

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_database \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --mongodb_conf hosts=127.0.0.1:27017 \
    --mongodb_conf username=root \
    --mongodb_conf password=123456 \
    --mongodb_conf database=source_db \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4

Example 2: synchronize selected collections

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_database \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --mongodb_conf hosts=127.0.0.1:27017 \
    --mongodb_conf username=root \
    --mongodb_conf password=123456 \
    --mongodb_conf database=source_db \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --including_tables 'product|user|address|order|custom'

Database Action Reference

Command syntax (square brackets indicate optional arguments):

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-2.2-SNAPSHOT.jar \
    mongodb_sync_database \
    --warehouse <warehouse-path> \
    --database <database-name> \
    [--table_prefix <paimon-table-prefix>] \
    [--table_suffix <paimon-table-suffix>] \
    [--including_tables <mongodb-table-name|name-regular-expr>] \
    [--excluding_tables <mongodb-table-name|name-regular-expr>] \
    [--partition_keys <partition_keys>] \
    [--primary_keys <primary-keys>] \
    [--mongodb_conf <mongodb-cdc-source-conf> [--mongodb_conf <mongodb-cdc-source-conf> ...]] \
    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
    [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]

Configuration	Description
--warehouse	The path to Paimon warehouse.
--database	The database name in Paimon catalog.
--table_prefix	The prefix of all Paimon tables to be synchronized. For example, if you want all synchronized tables to have "ods_" as prefix, you can specify "--table_prefix ods_".
--table_suffix	The suffix of all Paimon tables to be synchronized. The usage is same as "--table_prefix".
--including_tables	It is used to specify which source tables are to be synchronized. You must use '\|' to separate multiple tables.Quote the expression because '\|' is a shell operator, for example: 'a\|b\|c'. Regular expression is supported, for example, specifying "--including_tables test\|paimon.*" means to synchronize table 'test' and all tables start with 'paimon'.
--excluding_tables	It is used to specify which source tables are not to be synchronized. The usage is same as "--including_tables". "--excluding_tables" has higher priority than "--including_tables" if you specified both.
--partition_keys	The partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example "dt,hh,mm". If the keys are not in source table, the sink table won't set partition keys.
--primary_keys	The primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example "buyer_id,seller_id". If the keys are not provided, but the source has primary keys, the sink table will use source's primary keys. Otherwise, the sink table won't set primary keys. If the keys are not provided, but the source has primary keys, and you don't want to use source's primary keys, use --sync_primary_keys_from_source_schema.
--sync_primary_keys_from_source_schema	This is used to specify if primary keys from source should be used in paimon schema if primary keys using --primary_keys are not specified. The default is true.
--mongodb_conf	The configuration for Flink CDC MongoDB sources. Each configuration should be specified in the format "key=value". hosts, username, password, database are required configurations, others are optional. See its document for a complete list of configurations.
--catalog_conf	The configuration for Paimon catalog. Each configuration should be specified in the format "key=value". See here for a complete list of catalog configurations.
--table_conf	The configuration for Paimon table sink. Each configuration should be specified in the format "key=value". See here for a complete list of table configurations.

JSONPath Reference

Operators

Operator	Description
$	The root element to query. This starts all path expressions.
@	The current node being processed by a filter predicate.
*	Wildcard. Available anywhere a name or numeric are required.
..	Deep scan. Available anywhere a name is required.
.	Dot-notated child.
['{name}' (, '{name}')]	Bracket-notated child or children.
[{number} (, {number})]	Bracket-notated child or children.
[start:end]	Array index or indexes.
[?({expression})]	Filter expression. Expression must evaluate to a boolean value.

Functions

A function at the end of a path receives the result of that path expression.

Function	Description	Output type
min()	Provides the min value of an array of numbers.	Double
max()	Provides the max value of an array of numbers.	Double
avg()	Provides the average value of an array of numbers.	Double
stddev()	Provides the standard deviation value of an array of numbers	Double
length()	Provides the length of an array	Integer
sum()	Provides the sum value of an array of numbers.	Double
keys()	Provides the property keys (An alternative for terminal tilde ~)	Set
concat(X)	Provides a concatinated version of the path output with a new item.	like input
append(X)	add an item to the json path output array	like input
append(X)	add an item to the json path output array	like input
first()	Provides the first item of an array	Depends on the array
last()	Provides the last item of an array	Depends on the array
index(X)	Provides the item of an array of index: X, if the X is negative, take from backwards	Depends on the array

Path Examples

{
    "store": {
        "book": [
            {
                "category": "reference",
                "author": "Nigel Rees",
                "title": "Sayings of the Century",
                "price": 8.95
            },
            {
                "category": "fiction",
                "author": "Evelyn Waugh",
                "title": "Sword of Honour",
                "price": 12.99
            },
            {
                "category": "fiction",
                "author": "Herman Melville",
                "title": "Moby Dick",
                "isbn": "0-553-21311-3",
                "price": 8.99
            },
            {
                "category": "fiction",
                "author": "J. R. R. Tolkien",
                "title": "The Lord of the Rings",
                "isbn": "0-395-19395-8",
                "price": 22.99
            }
        ],
        "bicycle": {
            "color": "red",
            "price": 19.95
        }
    },
    "expensive": 10
}

JsonPath	Result
$.store.book[*].author	All book authors in the store.
$..author	All authors.
$.store.*	All things, both books and bicycles.
$.store..price	All prices in the store, including books and the bicycle.
$..book[2]	The third book.
$..book[-2]	The second to last book.
$..book[0,1]	The first two books.
$..book[:2]	All books from index 0 (inclusive) until index 2 (exclusive).
$..book[1:2]	All books from index 1 (inclusive) until index 2 (exclusive)
$..book[-2:]	Last two books
$..book[2:]	All books from index 2 (inclusive) to last
$..book[?(@.isbn)]	All books with an ISBN number
$.store.book[?(@.price < 10)]	All books in store cheaper than 10
$..book[?(@.price <= $['expensive'])]	All books in store that are not "expensive"
$..book[?(@.author =~ /.*REES/i)]	All books matching regex (ignore case)
$..*	Give me every thing
$..book.length()	The number of books

Prerequisites​

Synchronizing Tables​

Schema and Document Identity​

Example 1: synchronize collection into one Paimon table​

Example 2: Synchronize collection into a Paimon table according to the specified field mapping.​

Table Action Reference​

--warehouse

--database

--table

--partition_keys

--computed_column

--mongodb_conf

--catalog_conf

--table_conf

Synchronizing Databases​

Example 1: synchronize entire database​

Example 2: synchronize selected collections​

Database Action Reference​

--warehouse

--database

--table_prefix

--table_suffix

--including_tables

--excluding_tables

--partition_keys

--primary_keys

--sync_primary_keys_from_source_schema

--mongodb_conf

--catalog_conf

--table_conf

JSONPath Reference​

Operators​

$

@

*

..

.

['{name}' (, '{name}')]

[{number} (, {number})]

[start:end]

[?({expression})]

Functions​

min()

max()

avg()

stddev()

length()

sum()

keys()

concat(X)

append(X)

append(X)

first()

last()

index(X)

Path Examples​

$.store.book[*].author

$..author

$.store.*

$.store..price

$..book[2]

$..book[-2]

$..book[0,1]

$..book[:2]

$..book[1:2]

$..book[-2:]

$..book[2:]

$..book[?(@.isbn)]

$.store.book[?(@.price < 10)]

$..book[?(@.price <= $['expensive'])]

$..book[?(@.author =~ /.*REES/i)]

$..*

$..book.length()