Realtime Compute for Apache Flink：Kelola katalog MaxCompute

Buat tabel fisik MaxCompute menggunakan katalog

Saat Anda menggunakan pernyataan DDL Flink SQL untuk membuat tabel dalam katalog MaxCompute, tabel fisik yang sesuai akan dibuat secara otomatis di proyek MaxCompute yang sesuai. Tipe data Flink dikonversi secara otomatis ke tipe data MaxCompute. Anda dapat membuat tabel non-partisi dan tabel partisi.

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

Baca data dari tabel Katalog MaxCompute

Katalog MaxCompute dapat mengambil skema tabel fisik dari layanan MaxCompute. Oleh karena itu, Anda dapat langsung membaca data tanpa perlu mendeklarasikan skema yang sesuai di Flink. Contohnya:

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

Perilaku default saat tidak ada parameter yang dideklarasikan adalah membaca semua partisi. Untuk membaca partisi tertentu atau menggunakan mode tabel sumber inkremental, Anda dapat mendeklarasikan parameter dalam komentar SQL. Untuk informasi lebih lanjut tentang pengaturan parameter, lihat MaxCompute. Contohnya:

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;

Parameter lain untuk tabel sumber dan tabel dimensi yang didukung di MaxCompute dapat diatur dengan cara ini. Namun, perhatikan bahwa katalog MaxCompute tidak menyimpan informasi watermark. Untuk menentukan watermark saat membaca data dari tabel sumber, gunakan pernyataan CREATE TABLE ... LIKE .... Contohnya:

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

`ts` adalah kolom bertipe DATETIME dalam tabel fisik MaxCompute. Tipe ini dapat diatur sebagai event time di Flink, dan watermark dapat ditambahkan. Setelah tabel dibuat, semua data yang dibaca dari `newTable` memiliki watermark.

Tulis data ke tabel Katalog MaxCompute

Katalog MaxCompute mendukung penulisan data dalam mode partisi statis atau dinamis. Untuk informasi lebih lanjut, lihat MaxCompute. Misalnya, jika tabel fisik MaxCompute memiliki subpartisi `ds` dan `hh`, Anda dapat menggunakan pernyataan berikut untuk menulis data:

-- Tulis ke partisi statis
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=20231024,hh=09') */
SELECT <otherColumns>, '20231024', '09' FROM `<anotherTable>`;

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