AutoETL multi-cluster data aggregation - PolarDB - Alibaba Cloud Documentation Center

AutoETL supports aggregating business data from multiple PolarDB for MySQL clusters into a single PolarSearch node for unified full-text search and analytics. This topic describes how to create a multi-cluster data aggregation pipeline by using AutoETL.

Note

This feature is currently in canary release. To enable this feature, submit a ticket.

Background information

In real-world scenarios, business data may be distributed across multiple PolarDB for MySQL clusters. When you need to perform unified full-text search or analytics across these clusters, you can use the AutoETL multi-cluster data aggregation feature to synchronize data from multiple source clusters into a PolarSearch node on an aggregation cluster.

The data flow works as follows:

On each source cluster, run the CREATE ETL GRANT statement to authorize the aggregation cluster to access the data on the source cluster.
On the aggregation cluster, create an aggregation pipeline by using the ETL stored procedure (dbms_etl.sync_by_sql). In the WITH clause of each source table, specify the source PolarDB for MySQL cluster by using the polardb-mysql-instance parameter.
After the pipeline starts, the AutoETL pipeline verifies that each source cluster has granted ETL access to the aggregation cluster. After the authorization is confirmed, the engine reads data from each source cluster and synchronizes it to the PolarSearch node on the aggregation cluster.

Limits

All PolarDB for MySQL clusters involved in the aggregation must belong to the same Alibaba Cloud account (UID).
Multi-cluster aggregation pipelines can only be created by using the ETL stored procedure (dbms_etl.sync_by_sql). Search views are not supported.
Revoking an ETL grant does not affect synchronization pipelines that are already running.

Step 1: Create ETL grants

To allow the aggregation cluster to synchronize data from source clusters, you must create an ETL grant on each source cluster to authorize AutoETL access.

Create an ETL grant

Run the following SQL statement on the source cluster. Replace <allow_instance_id> with the cluster ID of the aggregation cluster.

CREATE ETL GRANT `<allow_instance_id>`;

Revoke an ETL grant

DROP ETL GRANT `<allow_instance_id>`;

View ETL grants

SHOW ETL GRANTS;

Step 2: Create an aggregation pipeline

You can create a multi-cluster aggregation pipeline by using the ETL stored procedure (dbms_etl.sync_by_sql). In the WITH clause of each source table, specify the source cluster ID by using the polardb-mysql-instance parameter.

Prepare data

Assume that you have two PolarDB for MySQL clusters (Cluster A and Cluster B), and you want to aggregate data from Cluster A into the PolarSearch node on Cluster B.

Create test data on Cluster A and grant ETL access to Cluster B:

-- Run on Cluster 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');

-- Grant ETL access to Cluster B
-- Replace pc-xxx with the actual cluster ID of Cluster B
CREATE ETL GRANT `pc-xxx`;

Create test data on Cluster B:

-- Run on Cluster 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');

Create the aggregation pipeline

Run the following SQL statement on Cluster B to create an aggregation pipeline that synchronizes data from both clusters to the PolarSearch node on Cluster B.

Note

Connection details for source and destination tables, such as connection addresses, ports, and credentials, are automatically configured by the system. You only need to specify the source cluster ID by using the polardb-mysql-instance parameter.

-- Run on Cluster B
CALL dbms_etl.sync_by_sql("search", "

-- Step 1: Define the source table on Cluster 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'  -- Replace with the actual cluster ID of Cluster A
);

-- Step 2: Define the source table on Cluster 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'  -- Replace with the actual cluster ID of Cluster B
);

-- Step 3: Define the PolarSearch destination table
CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `p1_c1` STRING,
  `p2_c1` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'dest'
);

-- Step 4: Define the computation and insertion logic
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`;
");

Step 3: Verify data

Connect to the PolarSearch node on the aggregation cluster and run the following Elasticsearch-compatible REST API query to verify that the data has been synchronized.

# Replace <user>:<password> with the credentials of the PolarSearch node
# Replace <polarsearch_endpoint> with the endpoint and port of the PolarSearch node
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

PolarDB:AutoETL multi-cluster data aggregation