Realtime Compute for Apache Flink:Manage MongoDB catalogs

Background information

This topic covers the following operations:

• Create a MongoDB catalog

• View a MongoDB catalog

• Use a MongoDB catalog

• Drop a MongoDB catalog

Before you begin

Confirm the following before working with MongoDB catalogs:

• Your Realtime Compute for Apache Flink workspace uses VVR 8.0.5 or later. VVR 8.0.6 or later is required if you plan to use the CREATE TABLE AS or CREATE DATABASE AS statement.

Limits

• Only Realtime Compute for Apache Flink with VVR 8.0.5 or later supports MongoDB catalogs.

• You cannot modify an existing MongoDB catalog using DDL statements. To update connection settings, drop the catalog and recreate it.

• MongoDB catalogs support read access only. You cannot create, modify, or delete databases and tables through a MongoDB catalog.

Create a MongoDB catalog

1. In the code editor of the Scripts tab on the SQL Editor page, enter the statement for creating a 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.enabled'='<flattenNestedColumns>',
    'scan.primitive-as-string'='<primitiveAsString>'
   );

   The following table describes the parameters.

   Parameter	Data type	Description	Required	Default value	Remarks
   yourcatalogname	STRING	The name of the MongoDB catalog.	Yes	N/A	Enter a custom name.
   type	STRING	The type of the catalog.	Yes	N/A	Set the value to mongodb.
   hosts	STRING	The hostname of the MongoDB instance.	Yes	N/A	Separate multiple hostnames with commas (,).
   default-database	STRING	The name of the default MongoDB database.	Yes	N/A	N/A.
   scheme	STRING	The connection protocol for the MongoDB database.	No	mongodb	Valid values: mongodb (uses the standard MongoDB protocol) mongodb+srv (uses the DNS SRV record protocol)
   username	STRING	The username for connecting to the MongoDB database.	No	N/A	Required if authentication is enabled on the MongoDB database.
   password	STRING	The password for connecting to the MongoDB database.	No	N/A	Required if authentication is enabled on the MongoDB database. To prevent credential exposure, use the key management feature instead of specifying a plaintext password. See Manage variables and keys.
   connection.options	STRING	Additional connection parameters for the MongoDB database.	No	N/A	Specify as key-value pairs in key=value format, separated by ampersands (&). For example: connectTimeoutMS=12000&socketTimeoutMS=13000.
   max.fetch.records	INT	The maximum number of documents the MongoDB catalog attempts to read when inferring a collection schema.	No	100	N/A.
   scan.flatten-nested-columns.enabled	BOOLEAN	Specifies whether to recursively expand nested columns when parsing BSON-formatted documents.	No	false	Valid values: true: Nested columns are expanded; the column path becomes the column name, for example {"nested": {"col": true}} becomes nested.col. false: Nested documents are parsed as STRING. Important This parameter applies only to MongoDB catalog tables used as source tables in Flink SQL deployments.
   scan.primitive-as-string	BOOLEAN	Specifies whether to infer all primitive data types as STRING when parsing BSON-formatted documents.	No	false	Valid values: true: All primitive types are inferred as STRING. false: Types are inferred from the BSON-to-Flink type mapping. For type mappings, see Schema inference description.

2. Select the code for creating the catalog and click Run on the left side of the code editor.

   

3. In the Catalogs pane on the left side of the Catalog List page, verify that the new catalog appears.

View a MongoDB catalog

1. In the code editor of the Scripts tab on the SQL Editor page, enter the following statement:

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

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

2. Select the code and click Run on the left side of the code editor.

   After the statement runs, the table details appear in the results panel.

   

Use a MongoDB catalog

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

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') */;

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;

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

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

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

Drop a MongoDB catalog

1. In the code editor of the Scripts tab on the SQL Editor page, enter the following statement:

   DROP CATALOG ${catalog_name};

   ${catalog_name} specifies the name of the MongoDB catalog that you want to drop.

2. Right-click the statement and choose Run from the shortcut menu to drop the catalog.

3. In the Catalogs pane on the left side of the Catalog List page, verify that the catalog no longer appears.

Schema inference description

When a MongoDB catalog infers the schema of a table, it automatically adds default parameters and primary key information. When the catalog parses BSON-formatted documents to obtain the schema of a collection, it attempts to read up to the number of data records specified by the max.fetch.records parameter. The catalog parses the schema of each record and merges all parsed schemas into a single collection schema.

A collection schema consists of the following parts:

• Physical columns: The MongoDB catalog infers physical columns from BSON-formatted documents.

• Default primary key: For any table in a MongoDB catalog, the _id column serves as the primary key to prevent duplicate data.

References

• For more information about how to use the MongoDB connector, see MongoDB connector.

• If the built-in catalogs of Realtime Compute for Apache Flink do not meet your requirements, you can use custom catalogs. For more information, see Manage custom catalogs.