Realtime Compute for Apache Flink:Gestionar catálogos de MongoDB

Contexto

Este tema describe cómo realizar las siguientes operaciones para gestionar un catálogo de MongoDB:

• Create a MongoDB catalog

• View a MongoDB catalog

• Use a MongoDB catalog

• Drop a MongoDB catalog

Límites

• Solo Ververica Runtime (VVR) 8.0.5 o versiones posteriores admiten catálogos de MongoDB.

• No puede usar una sentencia DDL para modificar un catálogo de MongoDB existente.

• Solo puede consultar tablas. No puede crear, modificar ni eliminar bases de datos ni tablas.

Crear un catálogo de MongoDB

1. En el editor de la pestaña Scripts, introduzca la sentencia para configurar el catálogo de MongoDB.

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

   Parámetro	Tipo	Descripción	Obligatorio	Observaciones
   yourcatalogname	String	El nombre del catálogo de MongoDB.	Sí	Especifique un nombre personalizado en inglés. Importante Después de reemplazar el parámetro con el nombre de su catálogo, debe eliminar los corchetes angulares (<>). De lo contrario, la comprobación de sintaxis fallará.
   type	String	El tipo de catálogo.	Sí	Establezca el valor en mongodb.
   hosts	String	El nombre de host del servidor de MongoDB.	Sí	Separe varios nombres de host con comas (,).
   default-database	String	El nombre de la base de datos predeterminada de MongoDB.	Sí	Ninguna.
   scheme	String	El protocolo de conexión utilizado para MongoDB.	No	Valores válidos: mongodb (predeterminado): se conecta mediante el protocolo estándar de MongoDB. mongodb+srv: se conecta mediante el protocolo de registro DNS SRV.
   username	String	El nombre de usuario utilizado para conectarse a MongoDB.	No	Este parámetro es obligatorio si la autenticación está habilitada.
   password	String	La contraseña utilizada para conectarse a MongoDB.	No	Este parámetro es obligatorio si la autenticación está habilitada. Nota Para evitar exponer la contraseña, se recomienda usar variables. Para más información, consulte Manage Variables.
   connection.options	String	Parámetros de conexión adicionales para el cliente de MongoDB.	No	Opciones adicionales en formato key=value separadas por signos et (&). Ejemplo: connectTimeoutMS=12000&socketTimeoutMS=13000.
   max.fetch.records	Int	El número máximo de documentos que se obtienen para inferir el esquema a partir de documentos BSON.	No	Valor predeterminado: 100.
   scan.flatten-nested-columns.enabled	Boolean	Especifica si se deben aplanar recursivamente los documentos anidados en BSON.	No	Valores válidos: true: Aplana recursivamente las columnas anidadas. Para una columna aplanada, Flink usa la ruta de acceso como nombre de columna. Por ejemplo, para la columna col en {"nested":{"col":true}}, el nombre de la columna aplanada es nested.col. false (predeterminado): trata los documentos BSON anidados como STRING. Importante Este parámetro solo es compatible cuando se utiliza una tabla del catálogo de MongoDB como tabla de origen en un despliegue de Flink SQL.
   scan.primitive-as-string	Boolean	Especifica si se deben inferir todos los tipos primitivos como STRING al analizar documentos BSON.	No	Valores válidos: true: infiere todos los tipos primitivos como STRING. false (predeterminado): infiere tipos según las reglas predeterminadas. Para más información, consulte Table details from a MongoDB catalog.

2. Seleccione la sentencia y haga clic en Run en el margen del editor.

   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. En el panel Catalogs de la izquierda, compruebe que aparece el nuevo catálogo.

Consultar un catálogo de MongoDB

1. En el editor de la pestaña Scripts, introduzca el siguiente comando.

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

   Parámetro	Descripción
   ${catalog_name}	El nombre del catálogo de MongoDB.
   ${db_name}	El nombre de la base de datos de MongoDB.
   ${collection_name}	El nombre de la colección de MongoDB.

2. Seleccione la sentencia y haga clic en Run en el margen del editor.

Usar un catálogo de MongoDB

• Como tabla de origen para leer datos de MongoDB.

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

  Nota

  Para especificar otras opciones WITH para una tabla de un catálogo de MongoDB, utilice un SQL hint. Por ejemplo, la sentencia SQL anterior usa un SQL hint para habilitar el escaneo paralelo para la instantánea inicial. Para más información sobre otras opciones WITH, consulte MongoDB.

• Como tabla de origen, use una CREATE TABLE AS (CTAS) statement o una CREATE DATABASE AS (CDAS) statement para sincronizar datos de MongoDB a una tabla de destino.

  Importante

  Para usar una sentencia CTAS o CDAS para sincronizar datos de MongoDB a una tabla de destino, se aplican los siguientes requisitos:

  • La versión de VVR debe ser 8.0.6 o posterior, y la versión de MongoDB debe ser 6.0 o posterior.

  • Los parámetros scan.incremental.snapshot.enabled y scan.full-changelog en el SQL hint deben establecerse en true.

  • La función de imágenes previas y posteriores debe estar habilitada para la base de datos de MongoDB. Para más información, consulte Document Preimages.

  • Sincronizar una sola tabla en tiempo real.

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

  • Sincronizar varias tablas en un solo despliegue.

    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;

    Puede usar un catálogo de MongoDB para sincronizar varias colecciones de MongoDB en un solo despliegue, siempre que se cumplan las siguientes condiciones:

    • Todas las tablas deben tener las mismas configuraciones de MongoDB, incluidos hosts, scheme, username, password y connection.options.

    • Todas las tablas deben tener la misma configuración de scan.startup.mode.

  • Sincronizar una base de datos completa.

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

• Leer datos de una tabla de dimensiones de MongoDB.

  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;

• Escribir datos de resultados en una tabla de MongoDB.

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

Después de que la sentencia se ejecute correctamente, puede ver los detalles de la tabla en los resultados de ejecución.

El esquema de la tabla incluye los siguientes campos: _id (STRING, clave principal, NOT NULL), name (STRING, anulable), age (INT, anulable) y addr (STRING, anulable).

Eliminar un catálogo de MongoDB

1. En el editor de la pestaña Scripts, introduzca el siguiente comando.

   DROP CATALOG ${catalog_name};

   En este comando, ${catalog_name} especifica el nombre del catálogo de MongoDB que se va a eliminar.

2. Seleccione la sentencia, haga clic derecho sobre ella y luego elija Run.

3. En el panel Catalogs de la izquierda, compruebe que el catálogo ya no aparece en la lista.

Detalles de tablas de un catálogo de MongoDB

Para simplificar su uso, un catálogo de MongoDB agrega automáticamente configuraciones predeterminadas y una clave principal a las tablas inferidas. Para inferir el esquema de una colección, el catálogo obtiene hasta max.fetch.records documentos, analiza cada uno y combina los resultados en un esquema final. El esquema incluye las siguientes partes:

• Columnas físicas inferidas

  Un catálogo de MongoDB infiere columnas físicas a partir de documentos BSON.

• Restricción de clave principal predeterminada

  Para las tablas de un catálogo de MongoDB, la columna _id se usa como clave principal de forma predeterminada para evitar datos duplicados.

Documentación relacionada

• Para más información sobre el conector de MongoDB, consulte MongoDB.

• Si los catálogos integrados no satisfacen sus requisitos empresariales, puede utilizar catálogos personalizados. Para más información, consulte Manage Custom Catalogs.