# 管理Iceberg Catalog
在AI DataLake Flink作业上配置Iceberg Catalog后，您可以通过AI DataLake Flink直接访问华为云数据湖构建（LakeFormation）中的Iceberg表。本文为您介绍如何在AI DataLake Flink作业中创建Iceberg Catalog，以及管理Iceberg数据库和Iceberg表。
**操作方法：** 使用Postman等工具，调用[创建FlinkJob作业](https://support.huaweicloud.com/api-aidatalake/RunFlinkJob.html)接口提交Flink作业。
#### 创建Iceberg LakeFormation Catalog
```
CREATE CATALOG lf_iceberg_catalog WITH (
  'type' = 'iceberg',
  'catalog-type' = 'lakeformation'
);
```
参数说明如下：
| 参数                          | 描述                                                                                                                      | 是否必填 | 备注                                            |
|:---|:---|:---|:---|
| type                       | Catalog类型。                                                                                                                  | 是   | 固定值为iceberg。                                  |
| catalog-type                | Catalog实现类型。                                                                                                               | 是    | 固定值为lakeformation。                          |
| lakecat.catalogname.default | 指定访问LakeFormation中的Catalog名称。 此名称需要与您在云数据湖构建（LakeFormation）中创建的Catalog名称保持一致，即可完成映射。 | 否    | 默认为Flink引擎侧的catalogName，即lf_iceberg_catalog。 |
   
#### 管理Iceberg数据库
- **创建数据库**
  创建Iceberg Catalog后，在lf_iceberg_catalog创建iceberg_db库。
  ```
  CREATE DATABASE IF NOT EXISTS lf_iceberg_catalog.iceberg_db
  WITH (
   'location' = 'obs://bucket/iceberg_catalog/iceberg_db/'
   );
  ```
  
- **删除数据库**
  ```
  DROP DATABASE lf_iceberg_catalog.iceberg_db; --只能删除内部没有任何表的数据库。
  ```
  
 
#### 管理Iceberg表
- 创建表 在名为iceberg-catalog的Catalog下的my_db的数据库中，创建一张分区表，其中分区键为dt，主键为dt、shop_id和user_id。代码示例如下。
  ```
  CREATE TABLE lf_iceberg_catalog.iceberg_db.iceberg_table (
    dt STRING,
    shop_id BIGINT,
    user_id BIGINT,
    num_orders INT,
    total_amount INT,
    PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
  ) PARTITIONED BY (dt);
  ```
  
- 修改表
  
  | **操作**      | **代码示例**                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
  |:---|:---|
  | 临时修改表参数 | 您可以通过在表名后添加SQL Hint的方式，在写入表时临时修改表参数。临时修改的表参数仅对当前SQL作业生效。 - 在写入my_table表时，临时将upsert-enabled设为true。 ``` INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */ SELECT ...; ```     - 在消费my_table表时，临时将monitor-interval设为10s。 ``` SELECT * FROM my_table /*+ OPTIONS('monitor-interval'='10s') */ ```    |
  | 重命名表       | 将表的名称从my_table修改为my_table_new。仅EXTERNAL表和LAKE表可修改。 ``` ALTER TABLE my_table RENAME TO my_table_new; ```                                                                                                                                                                                                                                                                                                                                                                                                                                                |
  | 添加新列     | 将类型为INT的c1列和类型为STRING的c2列添加在my_table表的末尾。 ``` ALTER TABLE my_table ADD (c1 INT, c2 STRING); ```                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
  | 重命名列      | 将my_table表中名称为c0的列重命名为c1。 ``` ALTER TABLE my_table RENAME c0 TO c1; ```                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
  | 删除列       | 将my_table表中c1和c2两列删除。 ``` ALTER TABLE my_table DROP (c1, c2); ```                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
  | 修改列注释     | 将my_table表中buy_count列的注释改为this is buy count。 ``` ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count'; ```                                                                                                                                                                                                                                                                                                                                                                                                                          |
  | 修改列顺序     | - 将my_table表中名称为col_a，类型为DOUBLE的列移动到开头。 ``` ALTER TABLE my_table MODIFY col_a DOUBLE FIRST; ```     - 将my_table表中名称为col_a，类型为DOUBLE的列移动到col_b列的后面。 ``` ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b; ```                                                                                                                                 |
     
  
- 删除表
  ```
  DROP TABLE lf_iceberg_catalog.iceberg_db.iceberg_table;
  ```