Elasticsearch:Use o Canal para sincronizar dados do MySQL com o Alibaba Cloud ES

Como funciona

O Canal atua como um nó secundário da sua instância do ApsaraDB RDS for MySQL e assina seus logs binários. À medida que linhas são inseridas, atualizadas ou excluídas no MySQL, o Canal captura cada evento de alteração e o grava no Elasticsearch por meio de uma API RESTful.

Pré-requisitos

Antes de começar, verifique se você tem:

• Uma instância do ApsaraDB RDS for MySQL. Consulte Criar uma instância do ApsaraDB RDS for MySQL. Este guia usa uma instância MySQL 5.7.

• Um cluster do Alibaba Cloud Elasticsearch. Consulte Criar um cluster do Alibaba Cloud Elasticsearch. Este guia usa um cluster Elasticsearch V6.7 (Kernel-enhanced Edition).

• Uma instância ECS com CentOS 7.6 (64 bits) para hospedar o Canal server e o Canal adapter. Consulte Criar uma instância usando o assistente.

• Todos os três recursos na mesma virtual private cloud (VPC).

• Endereço IP da instância ECS adicionado à lista de permissões de endereços IP do cluster do Elasticsearch. Consulte Configurar uma lista de permissões de endereços IP públicos ou privados para um cluster do Elasticsearch.

Limitações

• O Canal sincroniza apenas dados incrementais. Não há suporte para exportações completas de tabelas.

• O Canal adapter não se conecta a um cluster do Elasticsearch via HTTPS.

• Nomes dos campos e tipos de dados no mapeamento do índice do Elasticsearch devem corresponder exatamente às colunas da tabela MySQL.

• A versão do Java Development Kit (JDK) deve ser 1.8.0 ou superior.

• Compatibilidade de versões entre Canal e Elasticsearch:

  Nota

  Não use o Canal 1.1.4 para sincronizar dados com um cluster Elasticsearch V7.X. Use o Canal 1.1.5 para clusters Elasticsearch V7.x e o Canal 1.1.7 para clusters Elasticsearch V8.x. Este guia usa o Canal 1.1.4 com um cluster Elasticsearch V6.7.

  Versão do Canal	Versões suportadas do Elasticsearch
  1.1.4	6.x
  1.1.5	7.x
  1.1.7	8.x

Etapa 1: Preparar uma fonte de dados MySQL

Faça login no console do ApsaraDB RDS e crie um banco de dados e uma tabela. Este guia usa o seguinte esquema de tabela:

CREATE TABLE `es_test` (
    `id`    bigint(32) NOT NULL,
    `name`  text NOT NULL,
    `count` text NOT NULL,
    `color` text NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

Para instruções gerais, consulte Fluxo de trabalho geral para usar o ApsaraDB RDS for MySQL.

Etapa 2: Criar um índice do Elasticsearch com mapeamentos

Os nomes dos campos e tipos de dados no mapeamento do índice devem corresponder exatamente aos da tabela MySQL.

1. Faça login no console do Kibana do seu cluster do Elasticsearch. Consulte Fazer login no console do Kibana.

   Nota

   Este guia usa o Elasticsearch V6.7.0. As operações podem variar em outras versões.

2. No painel de navegação à esquerda, clique em Dev Tools.

3. Na aba Console, execute o seguinte comando para criar o índice es_test:

   PUT es_test?include_type_name=true
   {
     "settings": {
       "index": {
         "number_of_shards": "5",
         "number_of_replicas": "1"
       }
     },
     "mappings": {
       "_doc": {
         "properties": {
           "count": {
             "type": "text"
           },
           "id": {
             "type": "integer"
           },
           "name": {
             "type": "text",
             "analyzer": "ik_smart"
           },
           "color": {
             "type": "text"
           }
         }
       }
     }
   }

   Exemplo de resposta bem-sucedida:

   {
     "acknowledged": true,
     "shards_acknowledged": true,
     "index": "es_test"
   }

Etapa 3: Instalar o JDK

1. Conecte-se à instância ECS. Consulte Conectar-se a uma instância Linux usando senha ou chave.

   Nota

   Este exemplo usa um usuário comum.

2. Pesquise os pacotes JDK disponíveis:

   sudo yum search java | grep -i --color JDK

3. Instale o JDK. Este guia usa o java-1.8.0-openjdk-devel.x86_64:

   sudo yum install java-1.8.0-openjdk-devel.x86_64

4. Configure as variáveis de ambiente:

   1. Abra o arquivo ~/.bash_profile:

      vim ~/.bash_profile

   2. Adicione as seguintes linhas. Substitua JAVA_HOME pelo caminho real de instalação do seu JDK. Execute find / -name 'java' para localizá-lo.

      export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.362.b08-1.el7_9.x86_64
      export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
      export PATH=$PATH:$JAVA_HOME/bin

   3. Pressione Esc, digite :wq para salvar e aplique a configuração:

      source ~/.bash_profile

5. Verifique a instalação:

   java -version

   Saída esperada:

   openjdk version "1.8.0_362"
   OpenJDK Runtime Environment (build 1.8.0_362-b08)
   OpenJDK 64-Bit Server VM (build 25.362-b08, mixed mode)

Etapa 4: Instalar e iniciar o Canal server

1. Baixe o Canal server 1.1.4:

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz

   Para outras versões, consulte Releases do Canal.

2. Descompacte o pacote:

   tar -zxvf canal.deployer-1.1.4.tar.gz

3. Edite o arquivo conf/example/instance.properties:

   Parâmetro	Descrição	Exemplo
   canal.instance.master.address	Endpoint interno e porta da instância do ApsaraDB RDS for MySQL. Obtenha essas informações na página Basic Information da instância RDS.	rm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com:3306
   canal.instance.dbUsername	Nome de usuário do banco de dados. Obtenha esta informação na página Accounts da instância RDS.	canal_user
   canal.instance.dbPassword	Senha do banco de dados.	<your-password>

   vi conf/example/instance.properties

   Defina os seguintes parâmetros:

   

4. Pressione Esc, digite :wq para salvar.

5. Inicie o Canal server e verifique os logs:

   ./bin/startup.sh
   cat logs/canal/canal.log

   

Etapa 5: Instalar e iniciar o Canal adapter

1. Baixe o Canal adapter 1.1.4:

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz

2. Descompacte o pacote:

   tar -zxvf canal.adapter-1.1.4.tar.gz

3. Edite o arquivo conf/application.yml:

   Parâmetro	Descrição	Exemplo
   canal.conf.canalServerHost	Endereço do Canal server. Mantenha o valor padrão.	127.0.0.1:11111
   canal.conf.srcDataSources.defaultDS.url	URL de conexão JDBC da instância do ApsaraDB RDS for MySQL. Use o formato jdbc:mysql://<endpoint>:<port>/<database>?useUnicode=true. Obtenha o endpoint e a porta na página Basic Information.	jdbc:mysql://rm-bp1xxxxxxxxxnd6ph.mysql.rds.aliyuncs.com:3306/elasticsearch?useUnicode=true
   canal.conf.srcDataSources.defaultDS.username	Nome de usuário do banco de dados. Obtenha esta informação na página Accounts.	canal_user
   canal.conf.srcDataSources.defaultDS.password	Senha do banco de dados.	<your-password>
   canal.conf.canalAdapters.groups.outerAdapters.hosts (em name: es)	Endpoint interno e porta do cluster do Elasticsearch. Obtenha essas informações na página Basic Information do cluster.	es-cn-v64xxxxxxxxx3medp.elasticsearch.aliyuncs.com:9200
   canal.conf.canalAdapters.groups.outerAdapters.mode	Modo de transporte. Defina como rest.	rest
   canal.conf.canalAdapters.groups.outerAdapters.properties.security.auth	Credenciais do Elasticsearch no formato <username>:<password>.	elastic:es_password
   canal.conf.canalAdapters.groups.outerAdapters.properties.cluster.name	ID do cluster do Elasticsearch. Obtenha esta informação na página Basic Information.	es-cn-v64xxxxxxxxx3medp

   vi conf/application.yml

   Defina os seguintes parâmetros:

   

4. Pressione Esc, digite :wq para salvar.

5. Edite o arquivo de mapeamento de campos conf/es/*.yml para definir o mapeamento das colunas do MySQL para os campos do Elasticsearch:

   Parâmetro	Descrição	Valor usado neste guia
   esMapping._index	Nome do índice do Elasticsearch criado na Etapa 2.	es_test
   esMapping._type	Tipo do índice do Elasticsearch.	_doc
   esMapping._id	Campo de ID do documento.	_id
   esMapping.sql	Consulta SQL que seleciona as colunas a serem sincronizadas.	select t.id as _id, t.id, t.count, t.name, t.color from es_test t

   

6. Inicie o Canal adapter e verifique os logs:

   Nota

   O Canal adapter 1.1.4 inclui o driver MySQL versão 5.1.40. Se sua instância do ApsaraDB RDS executar o MySQL 8.x, o adapter falhará ao iniciar. Consulte

   Substituir o driver MySQL para MySQL 8.x

   na seção Solução de problemas.

   ./bin/startup.sh
   cat logs/adapter/adapter.log

   

Etapa 6: Verificar a sincronização

1. No banco de dados do ApsaraDB RDS for MySQL, insira uma linha na tabela es_test:

   INSERT INTO `ES`.`es_test` (`count`, `id`, `name`, `color`)
   VALUES ('11', 2, 'canal_test2', 'red');

   Também é possível testar com instruções UPDATE ou DELETE.

2. Faça login no console do Kibana do seu cluster do Elasticsearch. Consulte Fazer login no console do Kibana.

3. No painel de navegação à esquerda, clique em Dev Tools.

4. Na aba Console, execute a seguinte consulta:

   Importante

   O Canal sincroniza apenas alterações incrementais.

   GET /es_test/_search

   Se a sincronização funcionar corretamente, a resposta incluirá a linha inserida:

   

Solução de problemas

Os exemplos a seguir usam o usuário root.

java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassCastException:
com.alibaba.druid.pool.DruidDataSource cannot be cast to
com.alibaba.druid.pool.DruidDataSource
at com.alibaba.otter.canal.client.adapter.es7x.ES7xAdapter.init(ES7xAdapter.java:54)
~[client-adapter.es7x-1.1.5-jar-with-dependencies.jar:na]

Próximos passos

• Configurar uma lista de permissões de endereços IP públicos ou privados para um cluster do Elasticsearch

• Fazer login no console do Kibana

• Visualizar o status do cluster e informações dos nós

• Configurar sinônimos

• Plug-ins