Realtime Compute for Apache Flink:INSERT INTO ステートメント

背景情報

INSERT ステートメントでは、OPTIONS ヒントを使用して結果テーブルにパラメーターを渡すことができます。詳細については、「SQL ヒント」をご参照ください。

シンクにデータを書き込む方法の例

-- ソーステーブルを作成します。
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;

複数のシンクにデータを書き込む方法の例

この例では、データは 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;      -- 複数のシンクにデータを書き込む場合は、これが必要です。