Realtime Compute for Apache Flink:Manage MongoDB catalogs

Background information

This topic describes how to perform the following operations to manage a MongoDB catalog:

• Create a MongoDB catalog

• View a MongoDB catalog

• Use a MongoDB catalog

• Drop a MongoDB catalog

Limits

• Only Ververica Runtime (VVR) 8.0.5 or later supports MongoDB catalogs.

• You cannot use a DDL statement to modify an existing MongoDB catalog.

• You can only query tables. You cannot create, modify, or delete databases and tables.

Create a MongoDB catalog

1. In the editor on the Scripts tab, enter the statement to configure the MongoDB catalog.

   CREATE CATALOG <yourcatalogname> WITH(
    'type'='mongodb',
    'default-database'='<dbName>',
    'hosts'='<hosts>',
    'scheme'='<scheme>',
    'username'='<username>',
    'password'='<password>',
    'connection.options'='<connectionOptions>',
    'max.fetch.records'='100',
    'scan.flatten-nested-columns.enable'='<flattenNestedColumns>',
    'scan.primitive-as-string'='<primitiveAsString>'
   );

   Parameter	Type	Description	Required	Remarks
   yourcatalogname	String	The name of the MongoDB catalog.	Yes	Specify a custom name in English. Important After you replace the parameter with your catalog name, you must remove the angle brackets (<>). Otherwise, the syntax check fails.
   type	String	The type of the catalog.	Yes	Set the value to mongodb.
   hosts	String	The hostname of the MongoDB server.	Yes	Separate multiple hostnames with commas (,).
   default-database	String	The name of the default MongoDB database.	Yes	None.
   scheme	String	The connection protocol used for MongoDB.	No	Valid values: mongodb (default): connects using the standard MongoDB protocol. mongodb+srv: connects using the DNS SRV record protocol.
   username	String	The username used to connect to MongoDB.	No	This parameter is required if authentication is enabled.
   password	String	The password used to connect to MongoDB.	No	This parameter is required if authentication is enabled. Note To avoid exposing the password, we recommend using variables. For more information, see Manage Variables.
   connection.options	String	Additional connection parameters for the MongoDB client.	No	Extra options in the key=value format separated by ampersands (&). Example: connectTimeoutMS=12000&socketTimeoutMS=13000.
   max.fetch.records	Int	The maximum number of documents to fetch for schema inference from BSON documents.	No	Default value: 100.
   scan.flatten-nested-columns.enabled	Boolean	Specifies whether to recursively flatten nested documents in BSON.	No	Valid values: true: Recursively flattens nested columns. For a flattened column, Flink uses the access path as the column name. For example, for the col column in {"nested":{"col":true}}, the flattened column name is nested.col. false (default): treats nested BSON documents as STRING. Important This parameter is supported only when a table from the MongoDB catalog is used as a source table in a Flink SQL deployment.
   scan.primitive-as-string	Boolean	Specifies whether to infer all primitive types as STRING when parsing BSON documents.	No	Valid values: true: infers all primitive types as STRING. false (default): infers types based on the default rules. For more information, see Table details from a MongoDB catalog.

2. Select the statement and click Run in the gutter.

   CREATE CATALOG MongoDBCatalog WITH(
     'type'='mongodb',
     'default-database'='<dbName>',
     'hosts'='<hosts>',
     'scheme'='<scheme>',
     'username'='<username>',
     'password'='<password>',
     'connection.options'='<connectionOptions>',
     'max.fetch.records'='100',
     'scan.flatten-nested-columns.enable'='<flattenNestedColumns>',
     'scan.primitive-as-string'='<primitiveAsString>'
   );

3. In the Catalogs pane on the left, verify that the new catalog appears.

View a MongoDB catalog

1. In the editor on the Scripts tab, enter the following command.

   DESCRIBE `${catalog_name}`.`${db_name}`.`${collection_name}`;

   Parameter	Description
   ${catalog_name}	The name of the MongoDB catalog.
   ${db_name}	The name of the MongoDB database.
   ${collection_name}	The name of the MongoDB collection.

2. Select the statement and click Run in the gutter.

Use a MongoDB catalog

• As a source table to read data from MongoDB.

  INSERT INTO ${other_sink_table}
  SELECT...
  FROM `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
  /*+OPTIONS('scan.incremental.snapshot.enabled'='true')*/;

  Note

  To specify other WITH options for a table from a MongoDB catalog, use a SQL hint. For example, the preceding SQL statement uses a SQL hint to enable parallel scanning for the initial snapshot. For more information about other WITH options, see MongoDB.

• As a source table, use a CREATE TABLE AS (CTAS) statement or a CREATE DATABASE AS (CDAS) statement to synchronize data from MongoDB to a destination table.

  Important

  To use a CTAS or CDAS statement to synchronize data from MongoDB to a destination table, the following requirements apply:

  • The VVR version must be 8.0.6 or later, and the MongoDB version must be 6.0 or later.

  • The scan.incremental.snapshot.enabled and scan.full-changelog parameters in the SQL hint must be set to true.

  • The pre- and post-images feature must be enabled for the MongoDB database. For more information, see Document Preimages.

  • Synchronize a single table in real time.

    CREATE TABLE IF NOT EXISTS `${target_table_name}`
    WITH(...)
    AS TABLE `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
    /*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

  • Synchronize multiple tables in a single deployment.

    BEGIN STATEMENT SET;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table0`
    AS TABLE `mongodb-catalog`.`database`.`collection0`
    /*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table1`
    AS TABLE `mongodb-catalog`.`database`.`collection1`
    /*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;
    CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table2`
    AS TABLE `mongodb-catalog`.`database`.`collection2`
    /*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;
    END;

    You can use a MongoDB catalog to synchronize multiple MongoDB collections in a single deployment, provided that the following conditions are met:

    • All tables must have the same MongoDB configurations, including hosts, scheme, username, password, and connection.options.

    • All tables must have the same scan.startup.mode configuration.

  • Synchronize an entire database.

    CREATE DATABASE IF NOT EXISTS `some_catalog`.`some_database`
    AS DATABASE `mongodb-catalog`.`database`
    /*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

• Read data from a MongoDB dimension table.

  INSERT INTO ${other_sink_table}
  SELECT ...
  FROM ${other_source_table} AS e
  JOIN `${mongodb_catalog}`.`${db_name}`.`${table_name}` FOR SYSTEM_TIME AS OF e.proctime AS w
  ON e.id = w.id;

• Write result data to a MongoDB table.

  INSERT INTO `${mongodb_catalog}`.`${db_name}`.`${table_name}`
  SELECT ...
  FROM ${other_source_table}

After the statement runs successfully, you can view table details in the run results.

The table schema includes the following fields: _id (STRING, primary key, NOT NULL), name (STRING, nullable), age (INT, nullable), and addr (STRING, nullable).

Drop a MongoDB catalog

1. In the editor on the Scripts tab, enter the following command.

   DROP CATALOG ${catalog_name};

   In this command, ${catalog_name} specifies the name of the MongoDB catalog to drop.

2. Select the statement, right-click it, and then choose Run.

3. In the Catalogs pane on the left, verify that the catalog is no longer listed.

Table details from a MongoDB catalog

To simplify usage, a MongoDB catalog automatically adds default configurations and a primary key to inferred tables. To infer a collection's schema, the catalog fetches up to max.fetch.records documents, parses each one, and merges the results into a final schema. The schema includes the following parts:

• Inferred physical columns

  A MongoDB catalog infers physical columns from BSON documents.

• Default primary key constraint

  For tables from a MongoDB catalog, the _id column is used as the primary key by default to prevent duplicate data.

Related documentation

• For more information about the MongoDB connector, see MongoDB.

• If the built-in catalogs do not meet your business requirements, you can use custom catalogs. For more information, see Manage Custom Catalogs.