实时计算Flink版：管理MaxCompute Catalog

通过Catalog创建MaxCompute物理表

通过Flink SQL DDL，在MaxCompute Catalog中创建表时，会自动在对应的MaxCompute项目中创建对应的物理表，并自动将Flink中的类型转换为MaxCompute中的类型，支持创建非分区表和分区表。

CREATE TABLE `<catalogName>`.`<projectName>`.`<tableName>` (
 f0 INT,
  f1 BIGINT,
  f2 DOUBLE,
  f3 STRING
);

CREATE TABLE `<catalogName>`.`<projectName>`.`<tableName>` (
	f0 INT,
  f1 BIGINT,
  f2 DOUBLE,
  f3 STRING,
  ds STRING
) PARTITIONED BY (ds);

从MaxCompute Catalog表中读取数据

MaxCompute Catalog能够从MaxCompute服务读取物理表的Schema，因此无需在Flink中声明对应Schema即可直接读取数据。例如：

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`;

不声明任何参数的默认行为为全量读取所有分区，若您需要读取特定分区，或使用增量源表模式，可以参考大数据计算服务MaxCompute中的参数设置，在SQL注释中声明，例如：

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=230613') */;

SELECT * FROM `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('startPartition' = 'ds=230613') */;

SELECT * FROM `<anotherTable>` AS l LEFT JOIN
`<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'max_pt()', 'cache' = 'ALL') */
FOR SYSTEM_TIME AS OF l.proc_time AS r
ON l.id = r.id;

其他大数据计算服务MaxCompute中支持的源表和维表参数均可以通过该方式进行设置。但需要注意的是，MaxCompute Catalog中不保存Watermark信息，若需要在以源表读取数据时指定Watermark，可以使用CREATE TABLE ... LIKE ...语句，例如：

CREATE TABLE `<newTable>` ( WATERMARK FOR ts AS ts )
LIKE `<catalogName>`.`<projectName>`.`<tableName>`;

其中ts为MaxCompute物理表中类型为DATETIME的列，该类型可以在Flink中被设置为事件时间并添加Watermark信息，创建完成后，从newTable读取的数据均带有Watermark。

向MaxCompute Catalog表中写入数据

MaxCompute Catalog支持以固定分区或动态分区模式写入数据，参见结果表示例。例如有MaxCompute物理表有二级分区ds和hh，可以使用如下语句写入数据：

-- 写入固定分区
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=20231024,hh=09') */
SELECT <otherColumns>, '20231024', '09' FROM `<anotherTable>`;

-- 写入动态分区
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds,hh') */
SELECT <otherColumns>, ds, hh FROM `<anotherTable>`;