Elasticsearch:Usar o Logstash para sincronizar dados do ApsaraDB RDS for MySQL com o Elasticsearch

Limites

• Certifique-se de que a instância do ApsaraDB RDS for MySQL, o cluster do Alibaba Cloud Logstash e o cluster do Alibaba Cloud Elasticsearch estejam no mesmo fuso horário. Caso contrário, ao sincronizar dados relacionados a tempo, poderá haver diferença de fuso horário entre os dados antes e depois da sincronização.

• O campo _id no Elasticsearch deve ser idêntico ao campo id no MySQL.

  Essa condição garante que, quando um registro do MySQL for gravado no Elasticsearch, a tarefa de sincronização estabeleça um mapeamento direto entre o registro do MySQL e o documento do Elasticsearch. Por exemplo, ao atualizar um registro no MySQL, a tarefa de sincronização substitui o documento do Elasticsearch que possui o mesmo ID do registro atualizado.

  Nota

  Devido ao funcionamento interno do Elasticsearch, uma atualização consiste essencialmente na exclusão do documento antigo seguida pela indexação do novo documento. Portanto, substituir um documento no Elasticsearch tem a mesma eficiência de uma operação de atualização.

• Ao inserir ou atualizar dados no MySQL, o registro correspondente deve conter um campo que armazene o horário de atualização ou de inserção.

  Sempre que o Logstash consulta o MySQL, ele salva o horário de atualização ou inserção do último registro lido. Na leitura seguinte, o Logstash processa apenas os registros que atendem à condição, ou seja, aqueles cujo horário de atualização ou inserção é posterior ao do último registro da consulta anterior.

  Importante

  O plug-in logstash-input-jdbc não sincroniza exclusões. Execute o comando correspondente no Elasticsearch para excluir os documentos manualmente.

Pré-requisitos

Recomendamos criar as seguintes instâncias na mesma virtual private cloud (VPC):

Também é possível utilizar serviços implantados na Internet. Nesse caso, configure o SNAT, ative o endpoint público da instância do ApsaraDB RDS for MySQL e remova a restrição da lista de permissões. Para mais informações sobre como configurar o SNAT, consulte Configure a NAT gateway for public data transfer. Para detalhes sobre a configuração da lista de permissões, consulte Configure an IP address whitelist.

• Crie uma instância do ApsaraDB RDS for MySQL. Para mais informações, consulte Create an ApsaraDB RDS for MySQL instance. Neste tópico, usa-se o MySQL 5.7.

• Crie um cluster do Alibaba Cloud Elasticsearch. Para mais informações, consulte Create an Alibaba Cloud Elasticsearch cluster. Neste tópico, usa-se um cluster do Elasticsearch V8.17.

• Crie um cluster do Alibaba Cloud Logstash. Para mais informações, consulte Create an Alibaba Cloud Logstash instance. Neste tópico, usa-se um cluster do Logstash V8.11.4.

Informações básicas

O Alibaba Cloud Logstash é uma ferramenta de coleta e processamento de dados que oferece recursos de coleta, transformação, otimização e saída. O plug-in logstash-input-jdbc do Logstash vem instalado por padrão e não pode ser desinstalado. Com esse plug-in, você consulta dados em lotes numa instância do ApsaraDB RDS for MySQL e os sincroniza com o Elasticsearch. Ele também consulta periodicamente os dados da instância do ApsaraDB RDS for MySQL e sincroniza com o Elasticsearch os registros inseridos ou alterados desde a última consulta. Para mais informações, consulte "How to keep Elasticsearch synchronized with a relational database using Logstash" na documentação oficial. Essa solução é adequada para cenários em que se aceita uma latência de segundos na sincronização completa dos dados ou para consultas em lote de dados que atendem a condições específicas antes da sincronização.

Sincronizar dados

Perguntas frequentes

Meu pipeline está travado no estado de inicialização, os dados estão inconsistentes após a sincronização ou a conexão com o banco de dados falhou. O que devo fazer?

Verifique primeiro os logs do cluster do Logstash. Acesse o console do Logstash e use o recurso Query logs para visualizar os detalhes do erro. Para mais informações, consulte Query logs.

Se uma atualização do cluster estiver em andamento quando você aplicar uma correção, pause a atualização primeiro. Consulte View the progress of a cluster task . Após a correção, o sistema reiniciará o cluster e retomará a atualização automaticamente.

A tabela a seguir lista causas comuns e soluções:

Para mais orientações sobre solução de problemas, consulte FAQ about data transfer by using Logstash.

Como sincronizo dados de várias tabelas do MySQL para índices separados do Elasticsearch?

Defina múltiplos blocos jdbc na seção input, atribua um valor type a cada um e use condições if[type] na seção output para rotear os dados de cada tabela para um índice diferente:

input {
    jdbc {
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableA where update_time >= :sql_last_value"
      schedule => "* * * * *"
      record_last_run => true
      last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
      clean_run => false
      tracking_column_type => "timestamp"
      use_column_value => true
      tracking_column => "update_time"
      type => "A"
    }
    jdbc {
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableB where update_time >= :sql_last_value"
      schedule => "* * * * *"
      record_last_run => true
      last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
      clean_run => false
      tracking_column_type => "timestamp"
      use_column_value => true
      tracking_column => "update_time"
      type => "B"
    }
}
output {
    if[type] == "A" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_A"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
    if[type] == "B" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_B"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
}