通過Catalog創(chuàng)建MaxCompute物理表

通過Flink SQL DDL，在MaxCompute Catalog中創(chuàng)建表時(shí)，會自動(dòng)在對應(yīng)的MaxCompute項(xiàng)目中創(chuàng)建對應(yīng)的物理表，并自動(dòng)將Flink中的類型轉(zhuǎn)換為MaxCompute中的類型，支持創(chuàng)建非分區(qū)表和分區(qū)表。

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表中讀取數(shù)據(jù)

MaxCompute Catalog能夠從MaxCompute服務(wù)讀取物理表的Schema，因此無需在Flink中聲明對應(yīng)Schema即可直接讀取數(shù)據(jù)。例如：

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

不聲明任何參數(shù)的默認(rèn)行為為全量讀取所有分區(qū)，若您需要讀取特定分區(qū)，或使用增量源表模式，可以參考大數(shù)據(jù)計(jì)算服務(wù)MaxCompute中的參數(shù)設(shè)置，在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;

其他大數(shù)據(jù)計(jì)算服務(wù)MaxCompute中支持的源表和維表參數(shù)均可以通過該方式進(jìn)行設(shè)置。但需要注意的是，MaxCompute Catalog中不保存Watermark信息，若需要在以源表讀取數(shù)據(jù)時(shí)指定Watermark，可以使用CREATE TABLE ... LIKE ...語句，例如：

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

其中ts為MaxCompute物理表中類型為DATETIME的列，該類型可以在Flink中被設(shè)置為事件時(shí)間并添加Watermark信息，創(chuàng)建完成后，從newTable讀取的數(shù)據(jù)均帶有Watermark。

向MaxCompute Catalog表中寫入數(shù)據(jù)

MaxCompute Catalog支持以固定分區(qū)或動(dòng)態(tài)分區(qū)模式寫入數(shù)據(jù)，參見結(jié)果表示例。例如有MaxCompute物理表有二級分區(qū)ds和hh，可以使用如下語句寫入數(shù)據(jù)：

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

-- 寫入動(dòng)態(tài)分區(qū)
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds,hh') */
SELECT <otherColumns>, ds, hh FROM `<anotherTable>`;