After you configure a Hologres catalog, you can directly read Hologres metadata in the console of fully managed Flink without the need to manually register Hologres tables. Hologres catalogs improve the efficiency of job development and ensure data accuracy. This topic describes how to configure, view, and drop a Hologres catalog in the console of fully managed Flink.

Background information

Hologres catalogs provide metadata, such as databases, tables, partitions, and views. The metadata also includes functions and information that are required to access data stored in a database or other external systems. For more information, see Catalogs.

You can perform the following operations to manage Hologres catalogs:

Limits

  • Only Flink 1.13 and later allow you to configure Hologres catalogs.
  • You cannot modify the DDL statements that are related to catalogs.

Create a Hologres catalog

  1. Log on to the Realtime Compute for Apache Flink console.
  2. On the Fully Managed Flink tab, find the workspace that you want to manage and click Console in the Actions column.
  3. In the left-side navigation pane, click Draft Editor.
  4. In the upper-left corner of the Draft Editor page, click New. In the New Draft dialog box, select STREAM / SQL from the Type drop-down list.
  5. In the script editor, enter the following statement to create a Hologres catalog.
    CREATE CATALOG <catalogname> WITH (
      'type' = 'hologres',
      'endpoint' = '<endpoint>', 
      'username' = '<AccessKey>',
      'password' = '<AccessSecret>',
      'dbname' = '<dbname>'
    );
    catalogname indicates the name of the Hologres catalog. The name can contain only lowercase letters and digits. The name cannot contain uppercase letters or special characters, such as hyphens (-) and underscores (_). The following table describes the parameters in the WITH clause.
    Parameter Description Required
    type The type of the catalog. Set the value to hologres. Yes
    endpoint The endpoint of the Hologres instance.

    For more information, see Instance configurations.

    Yes
    username The AccessKey ID of the Alibaba Cloud account. Yes
    password The AccessKey secret of the Alibaba Cloud account. Yes
    dbname The name of the default Hologres database that you want to access. Yes
    ignore-non-persisted-options Specifies whether to ignore non-persistence options when you use a Hologres catalog to create a table that uses non-persistence options. Valid values:
    • true: The table can be created and all non-persistence options are ignored. This is the default value.
    • false: An error is returned, which indicates that the table fails to be created.
    Note The persistence of Hologres catalog table options allows the system to reacquire the consistent information that you defined in the DDL statement when the system reads the table information from the catalog again. Only the following persistence options are supported: endpoint, username, password, and dbname.
    No
  6. Click Execute.
    After the statement is executed, the message "Query has been executed" appears. If you want to drop the Hologres catalog, follow the instructions provided in Drop a Hologres catalog.
  7. On the left side of the Draft Editor page, click the Schemas tab.
  8. Click the Refresh icon icon to refresh and view the Hologres catalog that you created.
    Refresh icon

View a Hologres catalog

After you create a Hologres catalog, you can perform the following operations to view the metadata in the Hologres catalog.

  1. Log on to the Realtime Compute for Apache Flink console.
  2. On the Fully Managed Flink tab, find the workspace that you want to manage and click Console in the Actions column.
  3. In the left-side navigation pane, click Draft Editor.
  4. On the left side of the Draft Editor page, click the Schemas tab.
  5. In the top navigation bar, switch to the Hologres catalog that you want to view. In this example, the Hologres catalog named holo is used.
    holo
  6. View the databases, tables, and functions in the Hologres catalog.
    Table name
    Note If the public schema is used, the schema prefix is not added to a table name. In this case, only the table name is used.

Use a Hologres catalog

Note
  • If the public schema is used, ${table_name} is used instead of ${schema_name.table_name}. This is because the schema prefix is not added to the table name.
  • The update data in the table that is read by using a Hologres catalog can be consumed. After a Hologres table is read, the ignoredelete attribute is automatically set to false and the mutatetype attribute is automatically set to insertorupdate for the table. For more information about the ignoredelete and mutatetype attributes, see Merge data into a wide table.
  • Create a table in Hologres.
    • Execute the USE CATALOG HoloName statement to directly reference the information of the Hologres service.
      USE CATALOG ${catalog_name};
      CREATE TABLE `${db_name}`.`${schema_name.table_name}`(
        ...
       ) WITH (
         'connector' = 'hologres'
       );
      For more information about the USE syntax, see USE statements.
    • Directly reference the information of the Hologres service in DDL statements.
      CREATE TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` (
        ...
      ) WITH (
        'connector' = 'hologres'
      );
      Note
      • When you create a table for the registered Hologres service, you must set the connector parameter to hologres in the WITH clause. Other parameters such as endpoint are optional.
      • You are not allowed to directly add or modify parameters in the WITH clause for a Hologres table. You can use SQL hints to add or modify the parameters in the WITH clause in the INSERT statement. For more information about parameters in the WITH clause, see the "Parameters in the WITH clause" section in Create a Hologres source table, Create a Hologres result table, and Create a Hologres dimension table.
  • Read data from a Hologres table.
    INSERT INTO ${other_sink_table}
    SELECT ...
    FROM `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`
  • Import result data to a Hologres table.
    INSERT INTO `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`
    SELECT ... 
    FROM ${other_source_table}
  • Use the Hologres catalog that you created as the catalog of the destination store that is used in the CREATE TABLE AS statement.
    CREATE TABLE IF NOT `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`
    WITH (
      'connector' = 'hologres'
    ) AS TABLE ${other_source_table};
    To ensure that data can be written to Hologres when data is synchronized from the source table, the Hologres catalog is forced to change the schema of the result table in the following scenarios:
    • A column whose data type is DECIMAL is used as the primary key in the schema of the source table.

      Hologres does not support the primary keys of the DECIMAL type. Therefore, Hologres changes the data type of the column to BIGINT. If the change does not meet your business requirements, you can use the CREATE TABLE AS statement to convert the data type of the referenced column to the STRING type and recreate a primary key.

    • The schema of the source table contains the TIME, TIMESTAMP, or TIMESTAMP_LTZ type. The precision of these data types is greater than 6.

      The precision of time types supported by Hologres is 6. To ensure that data can be written to Hologres, fully managed Flink implicitly discards the parts that are higher than the highest precision supported by Hologres.

  • Use the Hologres catalog that you created as the catalog of the destination store that is used in the CREATE DATABASE AS statement.
    CREATE DATABASE IF NOT EXISTS `${catalog_name}`.`${db_name}`
    WITH (
      'sink.parallelism' = '5' -- Specify the parallelism for each result table. 
    ) AS DATABASE ${other_source_database};

    During data synchronization, you can declare the parameters of the result table in the WITH clause. When the job starts, these parameters are applied to the result tables to which data needs to be synchronized. For more information about the parameters that can be adjusted, see Create a Hologres result table.

    In addition to the parameters of the Hologres result table, the CREATE DATABASE AS statement allows you to specify the schemaname parameter in the WITH clause. This way, data can be synchronized to the specified schema of the destination Hologres database. The following table describes the schemaname parameter.
    Parameter Description Required Remarks
    schemaname The name of the schema. No Default value: public.

Drop a Hologres catalog

  1. Log on to the Realtime Compute for Apache Flink console.
  2. On the Fully Managed Flink tab, find the workspace that you want to manage and click Console in the Actions column.
  3. In the left-side navigation pane, click Draft Editor.
  4. In the upper-left corner of the Draft Editor page, click New. In the New Draft dialog box, select STREAM / SQL from the Type drop-down list.
  5. In the script editor, enter the following statement:
    DROP CATALOG ${catalog_name}
    catalog_name is the name of the Hologres catalog that you want to drop from the console of fully managed Flink.
    Note The drop operation does not affect the jobs that are running. However, jobs that are not published or that need to be suspended and then resumed are affected. Proceed with caution.
  6. Click Execute.
  7. On the left side of the Draft Editor page, click the Schemas tab.
  8. Click the Refresh icon icon to refresh the page and check whether the Hologres catalog is dropped.
    Refresh icon