Realtime Compute for Apache Flink:Manage MaxCompute catalogs

Create a MaxCompute physical table

Running a Flink DDL CREATE TABLE statement against a MaxCompute catalog automatically creates the corresponding physical table in MaxCompute and maps data types between both systems.

Non-partitioned table

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

Partitioned table

Place the partition key column at the end of the schema and declare it in PARTITIONED BY:

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

In this example, f0, f1, f2, and f3 are regular columns and ds is the partition key.

Read data from a MaxCompute catalog table

Because the catalog fetches schemas from MaxCompute automatically, no DDL declaration is required before reading.

Read all data (default)

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

By default, this reads all partitions of a partitioned table.

Read a specific partition

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

Read incrementally (incremental source table)

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

Use as a dimension table

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;

For all available options, see MaxCompute connector.

Add a watermark

MaxCompute catalogs do not carry watermark information. To specify event time when using a catalog table as a source, use CREATE TABLE ... LIKE ...:

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

Here ts is a column of the DATETIME type in the MaxCompute physical table. After this statement runs, reads from <newTable> include watermark information.

Write data to a MaxCompute catalog table

MaxCompute catalogs support writing to both static and dynamic partitions. For the full list of result table options, see MaxCompute connector.

Write to a static partition

-- Example: two-level partition (ds, hh), writing to a fixed partition
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=20231024,hh=09') */
SELECT <otherColumns>, '20231024', '09' FROM `<anotherTable>`;

Write to a dynamic partition

INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds,hh') */
SELECT <otherColumns>, ds, hh FROM `<anotherTable>`;

MaxCompute to Flink

Flink to MaxCompute