PolarDB:AutoETL マルチクラスターデータ集約

ETL 権限の作成

ソースクラスターで次の SQL ステートメントを実行します。<allow_instance_id> を集約クラスターのクラスター ID に置き換えてください。

CREATE ETL GRANT `<allow_instance_id>`;

ETL 権限の取り消し

DROP ETL GRANT `<allow_instance_id>`;

ETL 権限の表示

SHOW ETL GRANTS;

データの準備

2 つの PolarDB for MySQL クラスター (クラスター A とクラスター B) があり、両クラスターのデータを結合して、集約クラスターであるクラスター B の PolarSearch ノードに集約する場合を想定します。

• クラスター A でテストデータを作成し、クラスター B に ETL アクセス権限を付与します。

  -- クラスター A で実行
  CREATE DATABASE IF NOT EXISTS db1;
  USE db1;
  CREATE TABLE IF NOT EXISTS t1 (
      id INT PRIMARY KEY,
      c1 VARCHAR(100),
      c2 VARCHAR(100)
  );
  INSERT INTO t1(id, c1, c2) VALUES
  (1, '1', '1'),
  (2, '1', '1'),
  (3, '1', '1');
  
  -- クラスター B に ETL アクセスを許可
  -- pc-xxx をクラスター B の実際のクラスター ID に置き換えてください
  CREATE ETL GRANT `pc-xxx`;

• クラスター B でテストデータを作成します。

  -- クラスター B で実行
  CREATE DATABASE IF NOT EXISTS db2;
  USE db2;
  CREATE TABLE IF NOT EXISTS t2 (
      id INT PRIMARY KEY,
      c1 VARCHAR(100),
      c2 VARCHAR(100)
  );
  INSERT INTO t2(id, c1, c2) VALUES
  (1, '2', '2'),
  (2, '2', '2'),
  (3, '2', '2');

集約パイプラインの作成

クラスター B で次の SQL ステートメントを実行し、両方のクラスターからクラスター B 上の PolarSearch ノードにデータを同期する集約パイプラインを作成します。

-- クラスター B で実行
CALL dbms_etl.sync_by_sql("search", "

-- ステップ 1：クラスター A のソーステーブルを定義
CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   BIGINT,
  `c1`   STRING,
  `c2`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1',
  'polardb-mysql-instance' = 'pc-xxx'  -- クラスター A の実際のクラスター ID に置き換えてください
);

-- ステップ 2：クラスター B のソーステーブルを定義
CREATE TEMPORARY TABLE `db2`.`t2` (
  `id`   BIGINT,
  `c1`   STRING,
  `c2`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db2',
  'table-name' = 't2',
  'polardb-mysql-instance' = 'pc-xxx'  -- クラスター B の実際のクラスター ID に置き換えてください
);

-- ステップ 3：PolarSearch 宛先テーブルを定義
CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `p1_c1` STRING,
  `p2_c1` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'dest'
);

-- ステップ 4：計算と挿入ロジックを定義
INSERT INTO `dest`
SELECT
    `t1`.`id`,
    `t1`.`c1`,
    `t2`.`c1`
FROM `db1`.`t1` AS `t1`
LEFT JOIN `db2`.`t2` AS `t2`
  ON `t1`.`id` = `t2`.`id`;
");