通過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>`;