Realtime Compute for Apache Flink:MaxCompute catalogs

Create MaxCompute physical tables

When you create a table in a MaxCompute Catalog using a Flink SQL DDL statement, a corresponding physical table is automatically created in the specified MaxCompute project. Flink data types are automatically converted to their corresponding MaxCompute types. You can create both non-partitioned tables and partitioned tables.

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);

Read data from MaxCompute tables

MaxCompute Catalog can retrieve the schema of a physical table from the MaxCompute service. This allows you to read data directly without declaring the table schema in Flink. For example:

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

By default, if you do not specify any parameters, Flink reads all data from all partitions. If you need to read from specific partitions or use the incremental source mode, you can specify options in a SQL comment, as described in the MaxCompute documentation. For example:

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;

You can set other source table and dimension table parameters supported by MaxCompute in the same way. Note that MaxCompute Catalog does not store watermark information. If you need to specify a watermark when reading data from a source table, use a CREATE TABLE ... LIKE ... statement. For example:

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

In this example, ts is a DATETIME column in the MaxCompute physical table. In Flink, you can use this column as the event time and add a watermark. After the table is created, all data read from newTable contains watermarks.

Write data to MaxCompute tables

MaxCompute Catalog supports writing data in static or dynamic partition mode. For more information, see MaxCompute. For example, if a MaxCompute physical table has two partition levels, ds and hh, you can use the following statements to write data:

-- Write to a static 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>`;