Realtime Compute for Apache Flink:Katalog MaxCompute

Membuat tabel fisik MaxCompute

Saat Anda membuat tabel dalam katalog MaxCompute menggunakan pernyataan DDL Flink SQL, tabel fisik yang sesuai secara otomatis dibuat di proyek MaxCompute yang ditentukan. Tipe data Flink secara otomatis dikonversi ke tipe MaxCompute yang sesuai. Anda dapat membuat tabel non-partisi maupun 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);

Membaca data dari tabel MaxCompute

Katalog MaxCompute dapat mengambil skema tabel fisik dari layanan MaxCompute. Hal ini memungkinkan Anda membaca data secara langsung tanpa perlu mendeklarasikan skema tabel di Flink. Contohnya:

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

Secara default, jika Anda tidak menentukan parameter apa pun, Flink akan membaca semua data dari semua partisi. Jika Anda perlu membaca dari partisi tertentu atau menggunakan mode sumber inkremental, Anda dapat menentukan opsi dalam komentar SQL, seperti yang dijelaskan dalam dokumentasi 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;

Anda dapat mengatur parameter tabel sumber dan tabel dimensi lain yang didukung oleh MaxCompute dengan cara yang sama. Perhatikan bahwa Katalog MaxCompute tidak menyimpan informasi watermark. Jika Anda perlu 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>`;

Dalam contoh ini, ts adalah kolom DATETIME dalam tabel fisik MaxCompute. Di Flink, Anda dapat menggunakan kolom ini sebagai event time dan menambahkan watermark. Setelah tabel dibuat, semua data yang dibaca dari newTable akan berisi watermark.

Menulis data ke tabel MaxCompute

Katalog MaxCompute mendukung penulisan data dalam mode partisi statis atau dinamis. Untuk informasi selengkapnya, lihat MaxCompute. Misalnya, jika tabel fisik MaxCompute memiliki dua level partisi, ds dan hh, Anda dapat menggunakan pernyataan berikut untuk menulis 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>`;