Agregação de dados de múltiplos clusters com o AutoETL - PolarDB: Central de Documentação do Alibaba Cloud

O AutoETL permite agregar dados de negócios de vários clusters PolarDB for MySQL em um único nó PolarSearch para pesquisa de texto completo e análises unificadas. Este tópico descreve como criar um pipeline de agregação de dados de múltiplos clusters usando o AutoETL.

Nota

Este recurso está atualmente em versão canary. Para ativar este recurso, envie um chamado.

Informações de contexto

Em cenários reais, os dados de negócios podem estar distribuídos em vários clusters PolarDB for MySQL. Quando você precisa executar pesquisas de texto completo ou análises unificadas nesses clusters, use o recurso de agregação de dados de múltiplos clusters do AutoETL para sincronizar dados de vários clusters de origem em um nó PolarSearch no cluster de agregação.

O fluxo de dados funciona da seguinte forma:

Em cada cluster de origem, execute a instrução CREATE ETL GRANT para autorizar o cluster de agregação a acessar os dados no cluster de origem.
No cluster de agregação, crie um pipeline de agregação usando o procedimento armazenado de ETL (dbms_etl.sync_by_sql). Na cláusula WITH de cada tabela de origem, especifique o cluster PolarDB for MySQL de origem usando o parâmetro polardb-mysql-instance.
Após o início do pipeline, o AutoETL verifica se cada cluster de origem concedeu acesso de ETL ao cluster de agregação. Com a autorização confirmada, o mecanismo lê os dados de cada cluster de origem e os sincroniza com o nó PolarSearch no cluster de agregação.

Limites

Todos os clusters PolarDB for MySQL envolvidos na agregação devem pertencer à mesma conta Alibaba Cloud (UID).
Os pipelines de agregação de múltiplos clusters só podem ser criados por meio do procedimento armazenado de ETL (dbms_etl.sync_by_sql). Não há suporte para visualizações de pesquisa.
A revogação de uma concessão de ETL não afeta os pipelines de sincronização que já estão em execução.

Etapa 1: Criar concessões de ETL

Para permitir que o cluster de agregação sincronize dados dos clusters de origem, você deve criar uma concessão de ETL em cada cluster de origem para autorizar o acesso do AutoETL.

Crie uma concessão de ETL

Execute a seguinte instrução SQL no cluster de origem. Substitua <allow_instance_id> pelo ID do cluster de agregação.

CREATE ETL GRANT `<allow_instance_id>`;

Revogue uma concessão de ETL

DROP ETL GRANT `<allow_instance_id>`;

Visualize as concessões de ETL

SHOW ETL GRANTS;

Etapa 2: Criar um pipeline de agregação

Crie um pipeline de agregação de múltiplos clusters usando o procedimento armazenado de ETL (dbms_etl.sync_by_sql). Na cláusula WITH de cada tabela de origem, especifique o ID do cluster de origem usando o parâmetro polardb-mysql-instance.

Preparar dados

Considere que você tenha dois clusters PolarDB for MySQL (Cluster A e Cluster B) e deseje agregar dados do Cluster A no nó PolarSearch do Cluster B.

Crie dados de teste no Cluster A e conceda acesso de ETL ao 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`;

Crie dados de teste no 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');

Criar o pipeline de agregação

Execute a seguinte instrução SQL no Cluster B para criar um pipeline de agregação que sincroniza dados de ambos os clusters para o nó PolarSearch no Cluster B.

Nota

O sistema configura automaticamente os detalhes de conexão das tabelas de origem e destino, como endereços de conexão, portas e credenciais. Você só precisa especificar o ID do cluster de origem usando o parâmetro polardb-mysql-instance.

-- 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`;
");

Etapa 3: Verifique os dados

Conecte-se ao nó PolarSearch no cluster de agregação e execute a seguinte consulta de API REST compatível com Elasticsearch para verificar se os dados foram sincronizados.

# 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:Agregação de dados de múltiplos clusters com o AutoETL