INSERT INTO による 1 つまたは複数のシンクへのデータ書き込み - Flink

このトピックでは、`INSERT INTO` 文を使用して、1 つのジョブで 1 つ以上のシンクに書き込む方法について説明します。

背景情報

`INSERT` 文は、`OPTIONS` キーワードを使用した SQL ヒントをサポートしており、結果テーブルにパラメータを渡すことができます。詳細については、「SQL Hints」をご参照ください。

単一シンクへの書き込み例

-- ソーステーブルを作成します。
CREATE TEMPORARY TABLE datagen_source (
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'datagen' 
);

-- 結果テーブルを作成します。
CREATE TEMPORARY TABLE blackhole_sink(
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'blackhole' 
);

-- DML 文
INSERT INTO blackhole_sink SELECT UPPER(name), score FROM datagen_source;

複数シンクへの書き込み例

重要

複数シンクに書き込む文は、BEGIN STATEMENT SET; で始まり、END; で終わる必要があります。

この例では、2 つのシンクにデータを書き込みます。

-- ソーステーブルを作成します。
CREATE TEMPORARY TABLE datagen_source (
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'datagen'
);

-- 結果テーブル A を作成します。
CREATE TEMPORARY TABLE blackhole_sinkA(
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'blackhole' 
);

-- 結果テーブル B を作成します。
CREATE TEMPORARY TABLE blackhole_sinkB(
  name VARCHAR,
  score BIGINT
) WITH (
  'connector' = 'blackhole' 
);

-- DML 文
BEGIN STATEMENT SET;      -- 複数シンクへの書き込みに必要です。
INSERT INTO blackhole_sinkA 
  SELECT UPPER(name), sum(score) 
  FROM datagen_source 
  GROUP BY UPPER(name);
INSERT INTO blackhole_sinkB 
  SELECT LOWER(name), max(score) 
  FROM datagen_source 
  GROUP BY LOWER(name);
END;      -- 複数シンクへの書き込みに必要です。