Manage Hologres catalogs - Realtime Compute for Apache Flink

Prerequisites

You must have a dedicated Hologres instance that contains a database. For more information, see Create a database.

Limitations

Catalogs cannot be modified. To apply changes, you must delete the existing catalog and create a new one.
You must use a Hologres dedicated instance. Realtime Compute for Apache Flink does not support shared cluster instances because it can only access Hologres internal tables.

Create a Hologres catalog

Important

A catalog's configuration cannot be modified after creation. To make changes, drop the existing catalog and create a new one.

UI

If you need to set parameters that the Hologres connector supports, create the Hologres catalog using SQL.

Go to the Data Management page.
1. Log on to the Realtime Compute for Apache Flink console, and click Console in the Actions column of the target workspace.
2. Click Catalogs.
Click Create Catalog, select Hologres, and then click Next.

Configure the parameters.

Parameter	Description	Required	Notes
catalogname	The name of the Hologres catalog.	Yes	Only lowercase letters (a-z) and digits (0-9) are supported. Uppercase letters, hyphens (-), underscores (_), and other special characters are not allowed.
endpoint	The network endpoint of the Hologres instance.	Yes	If the Hologres instance and Flink workspace are in the same VPC, go to the Hologres console. On the details page of the target instance, obtain the network information for the Specified VPC from the Network Information section. For other network types, see Obtain the endpoint of a Hologres instance.
username	The username of a custom account, in the format `BASIC$<user_name>`. The AccessKey ID of an Alibaba Cloud account or a RAM user.	Yes	The configured user must have permissions to access the corresponding Hologres database. For more information, see Hologres permission model and user management. To learn how to obtain an AccessKey ID and AccessKey Secret, see Obtain an AccessKey pair. Important To prevent your AK or password from being leaked, we recommend that you use project variables to enter their values. For more information, see Project variables. A catalog created with a custom account displays only the databases to which the account has permissions. A catalog created with an AccessKey displays all databases in the instance.
password	The password of the custom account. The AccessKey Secret of the Alibaba Cloud account or RAM user.	Yes
dbname	The name of the Hologres database.	Yes	The database must already exist in the target Hologres instance. Otherwise, creating the catalog fails.

Click OK.

After the creation is complete, you can view the new Catalog under Catalogs.

SQL

In the editor on the Data Query page, enter the statement to create a Hologres catalog.

Syntax:

CREATE CATALOG <catalogname> WITH (
  'type' = 'hologres',
  'endpoint' = '<endpoint>', 
  'username' = '<AccessKey ID>',
  'password' = '<AccessKey Secret>',
  'dbname' = '<dbname>'
);

Examples:

Simple example

Real-time consumption example

CREATE CATALOG holocatalog WITH (
  'type' = 'hologres',
  'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80', 
  'username' = 'LTAI********************',
  'password' = '${secret_values.ak_holo}',
  'dbname' = 'holo_test'
);

Binlog must be enabled to support consuming binlog data.

CREATE CATALOG holocatalog WITH (
  'type' = 'hologres',
  'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80', 
  'username' = 'LTAI********************',
  'password' = '${secret_values.ak_holo}',
  'dbname' = 'holo_test',
  'binlog' = 'true', -- When you create the catalog, you can set WITH parameters for source, dimension, and sink tables. These parameters are then applied by default to tables accessed through this catalog.
  'cdcmode' = 'true',
  'connectionpoolname' = 'the_conn_pool',
  'table_property.binlog.level' = 'replica', -- You can also pass persisted Hologres table properties when creating the catalog. Then, binlog is enabled by default for new tables.
  'table_property.binlog.ttl' = '259200'
);

The following table describes the parameters.

Parameter	Description	Required	Notes
catalogname	The name of the Hologres catalog.	Yes	Only lowercase letters (a-z) and digits (0-9) are supported. Uppercase letters, hyphens (-), underscores (_), and other special characters are not allowed.
type	The catalog type.	Yes	The value must be `hologres`.
endpoint	The endpoint of the Hologres instance.	Yes	If the Hologres instance and the Flink workspace are in the same VPC, go to the Hologres management console and obtain the specified VPC network information from the Network Information section on the details page of the target instance. For other network types, see Obtain the endpoint of a Hologres instance.
username	The AccessKey ID of your Alibaba Cloud account or RAM user.	Yes	For more information, see Obtain an AccessKey pair. Note To prevent your AK information from being leaked, we recommend that you use a variable to specify the value for password. For more information, see Project variables. The user corresponding to the specified AccessKey must have permissions to access the Hologres database. For more information about database permissions, see Hologres permission model.
password	The AccessKey Secret of your Alibaba Cloud account or RAM user.	Yes
dbname	The name of the Hologres database.	Yes	The database must already exist in the target Hologres instance. Otherwise, creating the catalog fails.
ignore-non-persisted-options	Specifies the behavior when creating a table with non-persisted options through the catalog.	No	Valid values: `true` (default): The table is created, and all non-persisted options are ignored. `false`: An error is reported, and the table fails to be created. Note A persisted table option is one whose value, as defined in the DDL statement, is stored in the Hologres catalog and can be retrieved later. Currently, only the `endpoint`, `username`, `password`, and `dbname` options are persisted.
catalog.table.metadata-columns	When you use this catalog to specify a source table, you can add metadata columns from the Hologres binlog source table to the table schema as needed. Separate multiple metadata columns with a semicolon (;). Example: `hg_binlog_event_type;hg_binlog_timestamp_us`.	No	By default, no metadata columns are added. If you configure this parameter, the specified metadata columns are added to the returned table schema. These columns apply only to Hologres binlog source tables. Therefore, tables from this catalog can be used only as source tables, not as sink tables or dimension tables. Six types of metadata columns are supported. For more information, see Hologres binlog fields. Note This parameter is supported only in Ververica Runtime (VVR) 8.0.11 and later.
Other parameters supported by the Hologres connector	You can specify other parameters, including WITH options, when creating the catalog. These parameters then become the default settings for all tables accessed through this catalog.	No	To use this feature, set the `ignore-non-persisted-options` parameter to `true`.

After you enter the catalog creation statement, click Run in the upper-right corner.

Hologres catalog

After you configure a Hologres catalog, follow these steps to view its metadata.

Go to the Catalogs page.
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left-side navigation pane, click Catalogs.
On the Catalog List page, find your catalog by its Name and Type.

Click View to see the databases and tables in the catalog. If the schema is public, the table name appears without a schema prefix.

Use a Hologres Catalog

Notes:

If the schema is public, when you enter ${schema_name.table_name}, you can omit the schema name prefix and enter the table name directly as ${table_name}.
Tables in a Hologres Catalog can consume UPDATE data. For these tables, the ignoredelete property defaults to false and the mutatetype property defaults to insertorupdate. For more information about these properties, see Merge data into a wide table and update data partially.

Create a Hologres table

This example creates a table named holotable in the holodb database of the holocatalog catalog.

Note

When you create a table in a registered Hologres data service, the connector parameter is required in the WITH clause, and its value must be hologres. You can omit other parameters, such as endpoint.
You cannot directly add or modify supported WITH parameters in a Hologres table. To add or modify these parameters, use SQL hints in an INSERT statement.

UI

Go to the Catalogs page.
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. Click Catalogs.
Go to the target database.
1. In the Actions column of the target catalog, click View.
2. In the Actions column of the target database, click View.

Create the Hologres table.

Click Create Table.
On the Built-in tab, select the Hologres connector and click Next.

Enter the CREATE TABLE statement and configure the parameters. The following code is an example.

Syntax

Example

CREATE TABLE `${catalog_name}`.`${db_name}`.`${table_name}` (
 ...
) WITH (
 'connector' = 'hologres'
);

CREATE TABLE `holocatalog`.`holo_test`.`product` (
 id INT,
 name STRING
) WITH (
 'connector' = 'hologres'
);

Click OK.

SQL

In the editor on the Data Query page, enter the CREATE TABLE statement.

You can create a Hologres table in one of the following ways:

`USE CATALOG`

Use a USE statement to reference the Hologres Catalog.

Syntax

Example

USE CATALOG ${catalog_name};
CREATE TABLE `${db_name}`.`${schema_name.table_name}`(
  ...
 ) WITH (
   'connector' = 'hologres'
 );

USE CATALOG holocatalog;
CREATE TABLE `holodb`.`holotable` (
 id INT,
 name STRING
) WITH (
 'connector' = 'hologres'
);

DDL

Reference the Hologres Catalog in the DDL statement.

Syntax

Example

CREATE TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`(
  ...
 ) WITH (
   'connector' = 'hologres'
 );

CREATE TABLE `holocatalog`.`holodb`.`holotable` (
 id INT,
 name STRING
) WITH (
 'connector' = 'hologres'
);

You can also set physical table properties in the DDL statement.

CREATE TABLE `holocatalog`.`holodb`.`holotable` (
  id INT,
  name STRING
) WITH (
  'connector' = 'hologres',
  'table_property.orientation' = 'column',
  'table_property.distribution_key' = 'a',
  'table_property.clustering_key' = 'b:desc',
  'table_property.bitmap_columns' = 'a,b',
  'table_property.segment_key' = 'c',
  'table_property.time_to_live_in_seconds' = '86400',
  'table_property.binlog.level' = 'replica',
  'table_property.binlog.ttl' = '86400'
);

After you enter the statement, click Run in the upper-right corner to create the table.

When you create a Hologres table, you can specify table properties in the WITH clause. Some properties cannot be modified later. Setting these properties correctly improves data organization and query performance.

Physical table properties: The supported properties are consistent with those in Hologres, except that you must add the table_property. prefix. For more information about the parameters, see Overview of table creation and Subscribe to Hologres binlogs.

Parameter	Description	Example	Modifiable
table_property.orientation	Specifies the table's storage format.	'table_property.orientation' = 'row,column'	No
table_property.table_group	Specifies the table group.	'table_property.table_group' = 'table_group_xxx'
table_property.distribution_key	Specifies the distribution key.	'table_property.distribution_key' = 'a,b'
table_property.clustering_key	Specifies the clustering key.	'table_property.clustering_key' = 'a,b:desc'
table_property.event_time_column (formerly table_property.segment_key)	Specifies the segment key.	'table_property.event_time_column' = 'c,d' (or 'table_property.segment_key' = 'c,d')
table_property.bitmap_columns	Specifies the bitmap index.	'table_property.bitmap_columns' = 'a:on,b:off'	Yes
table_property.dictionary_encoding_columns	Specifies the dictionary encoding.	'table_property.dictionary_encoding_columns' = 'a:on,b:off,c:auto'
table_property.time_to_live_in_seconds	Specifies the time to live (TTL) of table data.	'table_property.time_to_live_in_seconds' = '864000'
table_property.binlog.level	Specifies whether to enable binlog.	'table_property.binlog.level' = 'replica'
table_property.binlog.ttl	Specifies the TTL of the binlog.	'table_property.binlog.ttl' = '86400'

Lenient mode parameter: To enable lenient mode when creating a table with a Hologres Catalog, set the enableTypeNormalization parameter.

Item	Description
Use case	In CTAS scenarios, you may need to adjust the precision of an existing field's data type (for example, from VARCHAR(10) to VARCHAR(20)) or change the data type (for example, from SMALLINT to INT).
Notes	You must enable lenient mode when you run the CTAS job for the first time. If you do not enable lenient mode on the first run, you must delete the downstream table and perform a stateless restart of the job for the change to take effect. After you enable lenient mode, if an upstream data type changes, the CTAS job continues to run as long as the new and original types share the same normalized type. Otherwise, the types are considered incompatible, and the CTAS job throws an exception.
Values	`false` (default): Creates a Hologres physical table based on standard type mapping. `true`: Lenient mode is enabled. When a Hologres physical table is created, a data type with higher precision is used after type normalization. The current type normalization rules are as follows: TINYINT, SMALLINT, INT, and BIGINT are normalized to BIGINT. CHAR, VARCHAR, and STRING are normalized to STRING. FLOAT and DOUBLE are normalized to DOUBLE. Other data types are created based on the original type mapping rules. For more information, see Type mapping.

Modify a Hologres table

A Hologres Catalog supports the following table modification operations:

Actions	Syntax and Example
Modify table properties	Only some table properties can be modified. For more information, see Create a Hologres table. Syntax: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` SET ( 'table_property.binlog.level' = 'replica', 'table_property.binlog.ttl' = '64700' ); Example: ALTER TABLE `holocatalog`.`holodb`.`holotable` SET ( 'table_property.binlog.level' = 'replica', 'table_property.binlog.ttl' = '64700' );
Rename a table	Syntax: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` RENAME TO `${catalog_name}`.`${db_name}`.`${schema_name.new_table_name}`; Example: ALTER TABLE `holocatalog`.`holodb`.`holotable` RENAME TO `holocatalog`.`holodb`.`new_holotable`;
Add a column	Syntax: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` ADD <column_name> <column_datatype> COMMENT '<column_comment>'; Example: ALTER TABLE `holocatalog`.`holodb`.`holotable` ADD address STRING COMMENT 'Address information';
Rename a column	Syntax: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` RENAME <old_column_name> TO <new_column_name>; Example: ALTER TABLE `holocatalog`.`holodb`.`holotable` RENAME address TO new_address;
Modify a column comment	Syntax: ALTER TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` MODIFY <column_name> <original_column_type> COMMENT '<new_column_comment>'; Example: ALTER TABLE `holocatalog`.`holodb`.`holotable` MODIFY new_address STRING COMMENT 'New address information';

Read and write Hologres tables

Read data from a Hologres table and write it to a sink table.

By default, Flink reads Hologres source tables in batch mode and does not process new data in real time. To stream data from Hologres in real time, use one of the following methods:

Configure the Hologres Catalog at creation: When you create the catalog by using SQL, you can enable binlog to consume binlog data. For more information, see Real-time consumption example. Then, read the Hologres data. The following code provides an example.

Syntax	Example
INSERT INTO ${other_sink_table} SELECT ... FROM `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`;	INSERT INTO sink_table SELECT id, name FROM `holocatalog`.`holodb`.`holotable`;

Change to stream mode by using a table hint. Add the /*+ OPTIONS('binlog'='true') */ hint to the SQL statement. The following code provides an example.
```
INSERT INTO sinktable
SELECT id, name
  FROM `holocatalog`.`holodb`.`holotable`
  /*+ OPTIONS ('binlog' = 'true') */;
```

Write data from a source table to a Hologres table.

Syntax	Example
INSERT INTO `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` SELECT ... FROM ${other_source_table}	INSERT INTO `holocatalog`.`holodb`.`holotable` SELECT id, name FROM source_table;

CTAS destination

Syntax

CREATE TABLE IF NOT EXISTS `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`
WITH (
  'connector' = 'hologres'
) AS TABLE ${other_source_table};

Example

CREATE TABLE IF NOT EXISTS `holocatalog`.`holodb`.`holotable`
WITH (
  'connector' = 'hologres'
) AS TABLE source_table;

CTAS allows you to set physical table properties in the WITH clause. These properties are set when the destination table is created. For more information about the supported table properties, see Create a Hologres table.

During data synchronization from a source, the Hologres Catalog may rewrite the schema of the destination table to ensure that data can be written to Hologres. This occurs in the following cases:

The source schema uses a column of the DECIMAL type as a primary key.

Because Hologres does not support DECIMAL as a primary key, it rewrites the column type to BIGINT by default. If this change does not meet your requirements, you can use a CTAS statement to cast the column to the STRING type and re-create the primary key.
The source schema contains a column of the TIME, TIMESTAMP, or TIMESTAMP_LTZ type with a precision greater than 6.

Hologres supports a precision of up to 6 for time types. To ensure that data can be written to Hologres, Flink implicitly discards the digits that exceed the maximum precision supported by Hologres.

CDAS destination

Syntax

CREATE DATABASE IF NOT EXISTS `${catalog_name}`.`${db_name}`
WITH (
  'sink.parallelism' = '5' -- Set the parallelism for each sink table.
) AS DATABASE ${other_source_database};

Example

CREATE DATABASE IF NOT EXISTS `holocatalog`.`holodb`
WITH (
  'sink.parallelism' = '5' -- Set the parallelism for each sink table.
) AS DATABASE source_database;

WITH parameter settings:

You can declare parameters for sink tables. When the job starts, these parameters are applied to the downstream tables that need to be synchronized. For more information about the supported parameters, see Hologres sink table.

You can specify a schemaname to synchronize data to a specific schema in the destination Hologres database. The following table describes the parameter.

Parameter	Description	Required	Default value
schemaname	The schema name.	No	public

You cannot set physical table properties because the WITH clause applies to all destination tables, and individual tables may require different properties.

If you need to set table properties, you must manually create the destination tables before you start the CDAS job. For more information about physical table properties, see Create a Hologres table.

Delete a Hologres catalog

Important

Deleting a Hologres catalog does not affect running jobs, but it does affect unpublished jobs or jobs that require pausing and resuming. Proceed with caution.

UI

Go to the Catalogs page.
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left navigation pane, click Catalogs.
On the Catalog List page, find the target catalog and click Delete in the Actions column.
In the confirmation dialog, click Delete.
Confirm that the target catalog no longer appears in the Catalogs section on the left.

Flink SQL

In the text editor on the Data Query page, enter the following command.
```
DROP CATALOG ${catalog_name}
```
Where ${catalog_name} is the name of the Hologres Catalog to delete, as displayed in the Realtime Compute for Apache Flink console.
Select the command, right-click, and then choose Run.
Confirm that the target catalog no longer appears in the Catalogs section on the left.

FAQ

Real-time consumption of Hologres data: What to do when a job reading from a Hologres source table using a Hologres Catalog completes automatically instead of running continuously?
Network connectivity issues: How do I find the endpoint of a Hologres instance?
Source and sink storage limitations: Why do I receive the error "CREATE TABLE ... AS TABLE ... statement requires target catalog ... implements org.apache.flink.table.catalog.CatalogTableProvider interface."?

Realtime Compute for Apache Flink:Manage Hologres catalogs

Prerequisites

Limitations

Create a Hologres catalog

UI

SQL

Hologres catalog

Use a Hologres Catalog

Create a Hologres table

UI

SQL

`USE CATALOG`

DDL

Modify a Hologres table

Read and write Hologres tables

CTAS destination

CDAS destination

Delete a Hologres catalog

UI

Flink SQL

FAQ

Related documents