Realtime Compute for Apache Flink:Manage a Fluss Catalog

Prerequisites

• A Fluss cluster is created. For more information, see Create a Fluss cluster.

• The Fluss cluster and the Flink workspace are in the same region and can reach each other over the network, either in the same VPC or with network connectivity configured.

Limits

• A catalog cannot be modified. To change it, delete the existing catalog and create it again.

• A catalog created with the password method connects to the Fluss cluster as the administrator. All jobs that use this catalog access data with administrator permissions, so data access cannot be isolated per user.

• A catalog created with the password-free method authenticates permissions based on the identity of the current user. When different users share the same catalog, each user can access only the data they are authorized for in Fluss.

Create a Fluss catalog

The way you create a catalog depends on the version of the Fluss cluster:

• Fluss cluster earlier than 0.9-ali-5.0: only the password method is supported. You must provide the administrator username and password.

• Fluss cluster 0.9-ali-5.0 or later: the password-free method is recommended. No username or password is required, and permissions are authenticated automatically based on the identity of the current user. The password method still works but is not recommended.

CREATE CATALOG fluss_catalog WITH (
  'type' = 'fluss',
  'bootstrap.servers' = '<bootstrap.servers>',
  'default-database' = 'fluss',
  'client.security.protocol' = 'SASL',
  'client.security.sasl.mechanism' = 'PLAIN',
  'client.security.sasl.username' = '<username>',
  'client.security.sasl.password' = '${secret_values.fluss_password}'
);

Create a custom catalog

Fluss iterates faster than Realtime Compute for Apache Flink. This topic provides the latest custom catalog package. You can use it to try out Fluss catalog features that have not been released with a VVR version.

A custom catalog can be created only by uploading a JAR file in the console. After it is created, you use it in the same way as a built-in Fluss catalog.

1. Download the custom catalog JAR file: fluss-ali-vvr-11-0.9-ali-catalog-6.0.jar

2. In the Create Catalog dialog box, click the Custom Catalog tab.

3. Click Create Custom Catalog Type, upload the downloaded JAR file, and then click Next.

4. After the JAR file is loaded, set Catalog Type to fluss-latest and click OK.

5. Select this catalog, and then click Next to create the custom catalog.

   CREATE CATALOG fluss_catalog WITH (
     'type' = 'fluss-latest',
     'bootstrap.servers' = '<bootstrap.servers>',
     'default-database' = 'fluss',
     'client.security.protocol' = 'SASL',
     'client.security.sasl.mechanism' = 'PLAIN',
     'client.security.sasl.username' = '<username>',
     'client.security.sasl.password' = '<password>'
   );

   Parameter	Description	Required	Remarks
   type	The catalog type.	Yes	Fixed value: fluss-latest.
   bootstrap.servers	The list of server addresses of the Fluss cluster.	Yes	You can view it on the cluster details page of the Fluss console.
   default-database	The name of the database to connect to by default.	No	The default value is fluss.
   client.security.protocol	The security authentication protocol.	Yes	Fixed value: SASL.
   client.security.sasl.mechanism	The SASL authentication mechanism.	Yes	Fixed value: PLAIN.
   client.security.sasl.username	The administrator username of the Fluss instance.	Yes	You can view them on the cluster details page of the Fluss console.
   client.security.sasl.password	The administrator password of the Fluss instance.	Yes

View a Fluss catalog

After the Fluss catalog is configured, perform the following steps to view Fluss metadata.

1. Log on to the Realtime Compute console.

2. In the Actions column of the target workspace, click Console.

3. In the left-side navigation pane, click Catalogs.

4. On the Catalog List page, view the catalog name and type.

5. Click View to view the databases and tables in the target catalog.

Use a Fluss catalog

CREATE TABLE `${catalog_name}`.`${db_name}`.`${table_name}` (
  ...
);

-- Create a primary key table
CREATE TABLE `fluss_catalog`.`fluss`.`product` (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

-- Create a log table
CREATE TABLE `fluss_catalog`.`fluss`.`orders` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
);

USE CATALOG fluss_catalog;
USE fluss;

CREATE TABLE product (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

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

INSERT INTO `fluss_catalog`.`fluss`.`product`
SELECT shop_id, user_id, num_orders, total_amount
FROM source_table;

INSERT INTO ${other_sink_table}
SELECT ...
FROM `${catalog_name}`.`${db_name}`.`${table_name}`;

INSERT INTO sink_table
SELECT shop_id, user_id, num_orders, total_amount
FROM `fluss_catalog`.`fluss`.`product`;

Delete a Fluss catalog

DROP CATALOG ${catalog_name};

References

• Write data to and consume data from Fluss tables: describes how to write data to a Fluss table and consume data from it.

• Manage custom catalogs: use a custom catalog if the built-in catalogs do not meet your business requirements.