Realtime Compute for Apache Flink:MaxCompute カタログの管理

カタログを使用した MaxCompute 物理テーブルの作成

Flink SQL DDL 文を使用して MaxCompute カタログにテーブルを作成すると、対応する MaxCompute プロジェクトに物理テーブルが自動的に作成されます。Flink のデータ型は、自動的に MaxCompute のデータ型に変換されます。非パーティションテーブルとパーティションテーブルを作成できます。

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 カタログテーブルからのデータ読み取り

MaxCompute カタログは、MaxCompute サービスから物理テーブルのスキーマを取得できます。したがって、Flink で対応するスキーマを宣言することなく、直接データを読み取ることができます。例：

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

パラメーターが宣言されていない場合、デフォルトではすべてのパーティションが読み取られます。特定のパーティションを読み取るか、増分ソーステーブルモードを使用するには、SQL コメントでパラメーターを宣言できます。パラメーター設定の詳細については、「MaxCompute」をご参照ください。例：

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;

MaxCompute でサポートされている他のソーステーブルおよびディメンションテーブルのパラメーターもこの方法で設定できます。ただし、MaxCompute カタログはウォーターマーク情報を保存しないことに注意してください。ソーステーブルからデータを読み取る際にウォーターマークを指定するには、CREATE TABLE ... LIKE ... 文を使用します。例：

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

`ts` は、MaxCompute 物理テーブルの DATETIME 型の列です。この型は Flink でイベント時間として設定でき、ウォーターマークを追加できます。テーブルが作成された後、`newTable` から読み取られるすべてのデータにウォーターマークが付きます。

MaxCompute カタログテーブルへのデータ書き込み

MaxCompute カタログは、静的または動的パーティションモードでのデータ書き込みをサポートしています。詳細については、「MaxCompute」をご参照ください。たとえば、MaxCompute 物理テーブルにサブパーティション `ds` と `hh` がある場合、次の文を使用してデータを書き込むことができます。

-- 静的パーティションへの書き込み
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds=20231024,hh=09') */
SELECT <otherColumns>, '20231024', '09' FROM `<anotherTable>`;

-- 動的パーティションへの書き込み
INSERT INTO `<catalogName>`.`<projectName>`.`<tableName>`
/*+ OPTIONS('partition' = 'ds,hh') */
SELECT <otherColumns>, ds, hh FROM `<anotherTable>`;