Realtime Compute for Apache 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>`;