PolarDB:ETL ストアドプロシージャ

データ準備

PolarDB for PostgreSQL で、次の SQL 文を実行してサンプルデータベースとテーブルを作成し、テストデータを挿入します。

CREATE SCHEMA IF NOT EXISTS db1;
CREATE TABLE IF NOT EXISTS db1.t1 (
    id INT PRIMARY KEY,
    c1 VARCHAR(100),
    c2 VARCHAR(100)
);
INSERT INTO db1.t1(id, c1, c2) VALUES
(1, 'apple', 'red'),
(2, 'banana', 'yellow'),
(3, 'grape', 'purple');

同期パイプラインの作成

dbms_etl.sync_by_table ストアドプロシージャを使用して、db1.t1 テーブルから PolarSearch ノード上の dest インデックスへの同期タスクを作成します。

構文

SELECT dbms_etl.sync_by_table('search', '<source_table>', '<sink_table>', '<column_list>');

パラメーター

制限

• ソーステーブルには、プライマリキーまたは一意キーが必要です。

• column_list を指定する場合、プライマリキーまたは一意キーを含める必要があります。

• 異なる同期パイプラインで同じソーステーブルまたは宛先テーブルを使用しないでください。

• パイプラインの作成後、ソーステーブルに追加した新しい列は自動的に同期されません。新しい列を同期するには、パイプラインを再作成する必要があります。

• 宛先インデックスにカスタム設定を使用したい場合は、同期パイプラインを作成する前に、PolarSearch ノードでインデックスを作成し、その設定を定義します。パイプライン作成時に宛先インデックスが存在しない場合、システムが自動的に作成します。

例

• db1.t1 テーブル全体を PolarSearch の dest インデックスに同期する：

  SELECT dbms_etl.sync_by_table('search', 'db1.t1', 'dest', '');

• db1.t1 テーブルの id 列と c1 列を dest1 インデックスに同期する：

  SELECT dbms_etl.sync_by_table('search', 'db1.t1', 'dest1', 'id, c1');

• db1.t1 テーブルの id 列と c1 列を dest2 インデックスに同期し、フィールド名のマッピングを指定する (id を d_id に、c1 を d_c1 にマッピングします)：

  SELECT dbms_etl.sync_by_table('search', 'db1.t1', 'dest2', 'd_id(id), d_c1(c1)');

データの検証

PolarSearch ノードに接続し、Elasticsearch 互換の REST API を使用してクエリを実行し、データが同期されているか確認します。

# <polarsearch_endpoint> を PolarSearch ノードのエンドポイントに置き換えます
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

データ準備

PolarDB for PostgreSQL で、次の SQL 文を実行してサンプルデータベースとテーブルを作成し、テストデータを挿入します。

CREATE SCHEMA IF NOT EXISTS db2;
CREATE SCHEMA IF NOT EXISTS db3;
CREATE SCHEMA IF NOT EXISTS db4;

CREATE TABLE IF NOT EXISTS db2.t2 (id INT PRIMARY KEY, c2 INT);
CREATE TABLE IF NOT EXISTS db3.t3 (id INT PRIMARY KEY, c3 VARCHAR(10));
CREATE TABLE IF NOT EXISTS db4.t4 (id INT PRIMARY KEY, c4 INT);

INSERT INTO db2.t2(id, c2) VALUES (1, 111), (2, 222), (4, 444);
INSERT INTO db3.t3(id, c3) VALUES (1, 'aaa'), (3, 'ccc'), (4, 'ddd');
INSERT INTO db4.t4(id, c4) VALUES (1, 11), (2, 22), (3, 33);

同期パイプラインの作成

dbms_etl.sync_by_map ストアドプロシージャを使用して、複数のテーブルからデータを JOIN し、PolarSearch ノードのインデックスに集約します。

構文

SELECT dbms_etl.sync_by_map(
    'search',
    '<columns_map>', -- 宛先インデックスのフィールドとソーステーブルのフィールドのマッピング
    '<join_fields>', -- テーブル間の JOIN キー
    '<join_types>',  -- JOIN タイプ (inner, left)
    '<filter>'       -- データフィルター条件
);

パラメーター

制限

• 同期対象のすべてのソーステーブルには、プライマリキーが必要です。

• この機能はストリーム処理を使用します。同期中は結果整合性のみが保証されます。

• 宛先インデックスの更新モードは、delete then insert です。

例

• 2 つのテーブルの INNER JOIN：db4.t4 と db2.t2 の間で id フィールドに対して INNER JOIN を実行し、t4.c4 と t2.c2 を dest インデックスの c4 フィールドと c2 フィールドに同期します。

  SELECT dbms_etl.sync_by_map(
    'search',
    'dest.c4(db4.t4.c4),dest.c2(db2.t2.c2)',
    'dest.id=db4.t4.id,db2.t2.id',
    'inner',
    ''
  );

• 複数のテーブルでの混合 JOIN とフィルター：3 つのテーブル db4.t4、db2.t2、db3.t3 を JOIN します。t4 と t2 の間で INNER JOIN を実行し、t4 と t3 の間で LEFT JOIN を実行します。その後、データは t4.c4 > 10 かつ t2.c2 < 100 の条件でフィルター処理されます。

  SELECT dbms_etl.sync_by_map(
    'search',
    'dest.c4(db4.t4.c4),dest.c2(db2.t2.c2),dest.c3(db3.t3.c3)',
    'dest.id=db4.t4.id,db2.t2.id,db3.t3.id',
    'inner,left',
    'db4.t4.c4 > 10 and db2.t2.c2 < 100'
  );

データの検証

PolarSearch ノードに接続し、Elasticsearch 互換の REST API を使用してクエリを実行し、データが同期されているか確認します。

# <polarsearch_endpoint> を PolarSearch ノードのエンドポイントに置き換えます
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

データ準備

PolarDB for PostgreSQL で、次の SQL 文を実行してサンプルデータベースとテーブルを作成し、テストデータを挿入します。

CREATE SCHEMA IF NOT EXISTS syncbysql_test;

CREATE TABLE IF NOT EXISTS syncbysql_test.test_source (
    id SERIAL PRIMARY KEY,
    polardb_document JSONB NOT NULL
);
INSERT INTO syncbysql_test.test_source (polardb_document) VALUES
  ('{
      "mailTemplateId": {"S": "TPL-001"},
      "title": {"S": "Welcome Registration Notification"},
      "createdAt": {"S": "2026-01-15 10:30:00"},
      "updatedAt": {"S": "2026-01-15 10:30:00"}
  }'),
  ('{
      "mailTemplateId": {"S": "TPL-002"},
      "title": {"S": "Order Confirmation Email"},
      "createdAt": {"S": "2026-02-01 14:20:00"},
      "updatedAt": {"S": "2026-02-10 09:15:00"}
  }'),
  ('{
      "mailTemplateId": {"S": "TPL-003"},
      "title": {"S": "Password Reset Reminder"},
      "createdAt": {"S": "2026-02-20 08:00:00"},
      "updatedAt": {"S": "2026-03-01 16:45:00"}
  }');

同期パイプラインの作成

複雑な変換、集約、または計算が必要なシナリオでは、dbms_etl.sync_by_sql ストアドプロシージャが Flink SQL 構文をサポートし、データ同期ロジックを定義します。

構文

SELECT dbms_etl.sync_by_sql('search', '<sync_sql>');

例

syncbysql_test の test_source テーブルから polardb_document JSON フィールドの値を PolarSearch に書き込みます。

システムは、SQL 文内のプレースホルダー {postgresql_host}、{postgresql_port}、{postgresql_user}、{postgresql_password}、{search_host}、{search_port}、{search_user}、および {search_password} を自動的に置換します。SQL 文では、これらの固定プレースホルダーを使用するだけで済みます。

select dbms_etl.sync_by_sql(
    'search',
    'CREATE TEMPORARY TABLE test_source_for_sync (
      polardb_document STRING
) WITH (
    ''schema-name'' = ''syncbysql_test'',    -- テーブルのスキーマ名に置き換えます
    ''database-name'' = ''testdb'',          -- テーブルが属するデータベース名に置き換えます
    ''connector''= ''postgres-cdc'',
    ''hostname'' = ''{pg_host}'',
    ''port'' = ''{pg_port}'',
    ''username'' = ''{pg_user}'',
    ''password'' = ''{pg_password}'',
    ''table-name'' = ''test_source'',        -- 同期するテーブル名に置き換えます
    ''slot.name'' = ''test_slot_3'',        -- 名前をカスタマイズします。この例では test_slot_3 を使用します。
    ''decoding.plugin.name'' = ''pgoutput''
);

CREATE TEMPORARY TABLE test_source_opensearch (
    mail_template_id STRING,
    title STRING,
    created_at STRING,
    updated_at STRING,
    PRIMARY KEY (mail_template_id) NOT ENFORCED
) WITH (
    ''connector'' = ''opensearch'',
    ''hosts'' = ''{search_host}'',
    ''index'' = ''test_source_opensearch'',
    ''username'' = ''{search_user}'',
    ''password'' = ''{search_password}''
);

INSERT INTO test_source_opensearch
SELECT
    -- インポートするプロパティに応じてこのセクションを変更します
    JSON_VALUE(polardb_document, ''$.mailTemplateId.S'') AS mail_template_id,
    JSON_VALUE(polardb_document, ''$.title.S'') AS title,
    JSON_VALUE(polardb_document, ''$.createdAt.S'') AS created_at,
    JSON_VALUE(polardb_document, ''$.updatedAt.S'') AS updated_at
FROM test_source_for_sync
WHERE JSON_VALUE(polardb_document, ''$.mailTemplateId.S'') IS NOT NULL; 
') AS sync_id;

データの検証

PolarSearch ノードに接続し、Elasticsearch 互換の REST API を使用してクエリを実行し、データが同期されているか確認します。

# <polarsearch_endpoint> を PolarSearch ノードのエンドポイントに置き換えます
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/test_source_opensearch/_search"