管理 Fluss Catalog

前提条件

• 已创建 Fluss 集群，具体操作请参见创建 Fluss 集群。

• Fluss 集群与 Flink 工作空间位于同一地域且网络互通（位于同一 VPC 或已配置网络连通）。

使用限制

• 不支持修改 Catalog。如需修改，需先删除已创建的 Catalog 再重新创建。

• 密码方式创建的 Catalog 使用管理员身份连接 Fluss 集群，所有使用该 Catalog 的作业均以管理员权限访问数据，无法按用户隔离数据访问权限。

• 免密方式创建的 Catalog 根据当前操作用户身份自动进行权限认证，不同用户使用同一个 Catalog 时，只能访问其在 Fluss 上被授权的数据。

创建 Fluss Catalog

根据 Fluss 集群版本的不同，创建 Catalog 的方式有所差异：

• Fluss 集群版本低于 0.9-ali-5.0：仅支持密码方式创建 Catalog，需要使用管理员用户名和密码。

• Fluss 集群版本为 0.9-ali-5.0 及以上：推荐使用免密方式创建 Catalog，无需输入用户名和密码，系统会根据当前操作用户身份自动完成权限认证。密码方式仍可用，但不推荐。

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

创建自定义 Catalog

Fluss 迭代快于实时计算 Flink。此处提供最新的自定义 Catalog 包，用于提前体验尚未随 VVR 版本发布的 Fluss Catalog 功能。

自定义 Catalog 仅支持通过控制台上传 JAR 包创建。创建后的使用方式与内置 Fluss Catalog 一致。

1. 下载自定义 Catalog JAR 包：fluss-ali-vvr-11-0.9-ali-catalog-6.0.jar

2. 在创建 Catalog对话框中，选择自定义 Catalog页签。

3. 单击创建自定义 Catalog 类型，上传已下载的 JAR 包，单击下一步。

4. 加载完成后，选择Catalog 类型为fluss-latest，单击确认。

5. 选择此 Catalog，单击下一步，创建自定义 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>'
   );

   参数	说明	是否必填	备注
   type	Catalog 类型。	是	固定值为 fluss-latest。
   bootstrap.servers	Fluss 集群的服务器地址列表。	是	可通过 Fluss 控制台的集群详情进行查看。
   default-database	默认连接的数据库名称。	否	默认值为 fluss。
   client.security.protocol	安全认证协议。	是	固定值为 SASL。
   client.security.sasl.mechanism	SASL 认证机制。	是	固定值为 PLAIN。
   client.security.sasl.username	Fluss 实例的管理员用户名。	是	可通过 Fluss 控制台的集群详情进行查看。
   client.security.sasl.password	Fluss 实例的管理员密码。	是

查看 Fluss Catalog

Fluss Catalog 配置成功后，按以下步骤查看 Fluss 元数据。

1. 登录实时计算控制台。

2. 单击目标工作空间操作列下的控制台。

3. 在左侧导航栏，单击数据管理。

4. 在 Catalog 列表页面，查看 Catalog 名称和类型。

5. 单击查看，查看目标 Catalog 下的数据库和表。

使用 Fluss Catalog

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

-- 创建主键表
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 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`;

删除 Fluss Catalog

DROP CATALOG ${catalog_name};

相关文档

• Fluss 表数据写入和消费：介绍如何向 Fluss 表中写入数据和从 Fluss 表中消费数据。

• 管理自定义 Catalog：如果内置的 Catalog 无法满足业务需求，可使用自定义 Catalog。