Sincronizar dados do PolarDB-X (DRDS) para o Elasticsearch usando Logstash - Elasticsearch: Central de Documentação do Alibaba Cloud

Informações básicas

O Alibaba Cloud Logstash é uma ferramenta robusta para coleta e processamento de dados, capaz de ingerir, transformar, otimizar e enviar informações. O plugin logstash-input-jdbc, instalado por padrão e não removível, permite consultar dados do PolarDB-X em lotes e sincronizá-los com o Elasticsearch. Esse plugin também executa consultas periódicas de polling no PolarDB-X para sincronizar registros inseridos ou alterados desde a última verificação. Essa solução é ideal tanto para a sincronização completa inicial quanto para atualizações incrementais subsequentes, que geralmente apresentam latência de poucos segundos.

Pré-requisitos

Crie uma instância do PolarDB-X 1.0, um cluster do Alibaba Cloud Elasticsearch e um cluster do Alibaba Cloud Logstash. Além disso, crie um banco de dados na instância do PolarDB-X 1.0. Recomendamos criar a instância do PolarDB-X 1.0, o cluster do Elasticsearch e o cluster do Logstash na mesma Virtual Private Cloud (VPC).

Crie uma instância do PolarDB-X 1.0 e um banco de dados. Para mais informações, consulte Create a PolarDB-X 1.0 instance.
Para saber como criar um cluster do Alibaba Cloud Elasticsearch, consulte Create an Alibaba Cloud Elasticsearch cluster. Neste exemplo, utiliza-se um cluster Elasticsearch V6.7 da Standard Edition.
Para detalhes sobre a criação de um cluster do Alibaba Cloud Logstash, consulte Create an Alibaba Cloud Logstash cluster.

Nota
Caso precise usar o Logstash para coletar dados da Internet ou transferir dados coletados para a Internet, configure um gateway NAT (Network Address Translation) e utilize-o para conectar seu cluster do Logstash à Internet. Para mais informações, consulte Configure a NAT gateway for data transmission over the Internet.

Limites

Os valores do campo _id no cluster do Elasticsearch devem ser idênticos aos valores do campo id no banco de dados PolarDB-X 1.0.

Essa condição garante que a tarefa de sincronização de dados estabeleça um mapeamento entre os registros no banco de dados PolarDB-X 1.0 e os documentos no cluster do Elasticsearch. Ao atualizar um registro no banco de dados PolarDB-X 1.0, a tarefa de sincronização usa esse registro atualizado para sobrescrever o documento correspondente no cluster do Elasticsearch.

Nota
Na prática, uma operação de atualização no Elasticsearch exclui o documento original e indexa o novo. Portanto, a operação de sobrescrita tem a mesma eficiência de uma operação de atualização executada pela tarefa de sincronização.
Ao inserir ou atualizar um registro no banco de dados PolarDB-X 1.0, o registro deve conter um campo que indique o momento da inserção ou atualização.

A cada ciclo de polling do plugin logstash-input-jdbc, o sistema registra o horário da última inserção ou atualização processada. O Logstash sincroniza apenas os registros do PolarDB-X 1.0 cujo timestamp de inserção ou atualização seja posterior ao timestamp do último registro processado no ciclo anterior.

Importante
Se você excluir registros no banco de dados PolarDB-X 1.0, o plugin logstash-input-jdbc não conseguirá remover automaticamente os documentos correspondentes do cluster do Elasticsearch. Para excluir esses documentos, execute o comando apropriado diretamente no cluster do Elasticsearch.

Procedimento

Etapa 1: Preparação

Crie uma tabela na instância do PolarDB-X 1.0 e insira dados de teste.

Neste exemplo, usa-se a seguinte instrução para criar a tabela:

CREATE table food(
  id int PRIMARY key AUTO_INCREMENT,
  name VARCHAR (32),
  insert_time DATETIME,
  update_time DATETIME
);

As instruções abaixo inserem dados na tabela:

INSERT INTO food values(null,'chocolate',now(),now());
INSERT INTO food values(null,'yogurt',now(),now());
INSERT INTO food values(null,'ham sausage',now(),now());

Ative o recurso Auto Indexing no cluster do Elasticsearch. Para mais informações, consulte Access and configure an Elasticsearch cluster.
Faça upload de um driver JDBC SQL compatível com a versão do banco de dados PolarDB-X 1.0 no cluster do Logstash. Para mais informações, consulte Configure third-party libraries. Neste exemplo, utiliza-se o driver mysql-connector-java-5.1.35.

Nota
Este exemplo utiliza um driver JDBC MySQL para conectar-se ao banco de dados PolarDB-X 1.0. Também é possível usar um driver JDBC do PolarDB. No entanto, para bancos de dados PolarDB-X 2.0, o driver JDBC do PolarDB pode não funcionar corretamente. Recomendamos o uso do driver JDBC MySQL.
Obtenha os endereços IP dos nós do cluster do Logstash na página Basic Information do console do Elasticsearch. Em seguida, adicione esses endereços à lista de permissões de IP da instância do PolarDB-X 1.0. Para mais informações, consulte Set an IP address whitelist.

Etapa 2: Configurar um pipeline do Logstash

Acesse a página Logstash Clusters.
Acesse o cluster desejado.
1. Na barra de navegação superior, selecione a região onde o cluster está localizado.
2. Na página Logstash Clusters, localize o cluster e clique em seu ID.
No painel de navegação à esquerda, clique em Pipelines.
Clique em Create Pipeline.

Na página Create, insira um Pipeline ID e configure o pipeline na caixa de texto Config.

Neste exemplo, inserem-se as seguintes configurações no campo Config Settings:

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.35.jar"
    jdbc_connection_string => "jdbc:mysql://drdshbga51x6****.drds.aliyuncs.com:3306/<Your database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    statement => "select * from food 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"
  }
}
filter {

}
output {
  elasticsearch {
    hosts => "http://es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com:9200"
    user => "elastic"
    password => "es_password"
    index => "drds_test"
    document_id => "%{id}"
  }
}

Nota

No código, substitua <Logstash cluster ID> pelo ID do seu cluster do Logstash. Para mais informações, consulte View cluster list overview.

Tabela 1. Parâmetros na parte de entrada (input)

Parâmetro	Descrição
jdbc_driver_class	Classe do driver JDBC.
jdbc_driver_library	Caminho do arquivo do driver JDBC. Para mais informações, consulte Configure third-party libraries.
jdbc_connection_string	String de conexão JDBC usada para acessar o banco de dados PolarDB-X 1.0. Inclui o endpoint, número da porta e nome do banco de dados.
jdbc_user	Nome de usuário para acesso ao banco de dados PolarDB-X 1.0.
jdbc_password	Senha para acesso ao banco de dados PolarDB-X 1.0.
jdbc_paging_enabled	Define se a paginação será ativada. Valor padrão: false.
jdbc_page_size	Quantidade de registros por página.
statement	Instrução SQL a ser executada.
schedule	Cronograma da tarefa de sincronização. `"* * * * *"` indica sincronização a cada minuto.
record_last_run	Indica se o resultado da última execução deve ser registrado. Se definido como true, o valor de tracking_column da última execução é salvo no arquivo especificado pelo parâmetro last_run_metadata_path.
last_run_metadata_path	Caminho do arquivo que armazena o horário da última execução. O backend fornece um caminho no formato /ssd/1/<Logstash cluster ID>/logstash/data/. Após definir o caminho, o Logstash gera automaticamente o arquivo, mas não é possível visualizar seu conteúdo.
clean_run	Define se o caminho especificado em last_run_metadata_path deve ser limpo. Valor padrão: false. Se true, cada consulta reinicia a partir do primeiro registro do banco de dados.
use_column_value	Indica se os valores de uma coluna específica devem ser rastreados.
tracking_column_type	Tipo da coluna cujos valores serão rastreados. Valor padrão: numeric.
tracking_column	Coluna cujos valores serão rastreados. Os valores devem estar em ordem crescente. Geralmente, trata-se da chave primária da tabela.

Tabela 2. Parâmetros na parte de saída (output)

Parâmetro	Descrição
hosts	Endpoint privado do cluster do Alibaba Cloud Elasticsearch, no formato `http://<Private endpoint of the cluster>:9200`. Encontre o endpoint privado na página Basic information. Para mais informações, consulte View the basic information of a cluster.
user	Nome de usuário para acessar o cluster do Elasticsearch. O padrão é elastic.
password	Senha da conta elastic. Essa senha é definida durante a criação do cluster do Elasticsearch. Caso tenha esquecido, redefina-a. Para detalhes sobre o procedimento e precauções, consulte Reset the access password for an Elasticsearch cluster.
index	Nome do índice no cluster do Elasticsearch.
document_id	IDs dos documentos no cluster do Elasticsearch. Defina como %{id} para garantir que os IDs dos documentos correspondam aos IDs dos registros no banco de dados PolarDB-X 1.0.

Importante

As configurações acima baseiam-se em dados de teste. Adapte o pipeline conforme suas necessidades de negócio. Para conhecer outros parâmetros suportados pelo plugin de entrada, consulte Plugin de entrada Jdbc do Logstash.
Se a configuração incluir parâmetros como last_run_metadata_path, o caminho do arquivo deve ser fornecido pelo service Alibaba Cloud Logstash. O sistema disponibiliza o caminho /ssd/1/<Logstash cluster ID>/logstash/data/ para testes. Os dados nesse diretório não são excluídos automaticamente; certifique-se de ter espaço em disco suficiente. O Logstash cria o arquivo no caminho especificado, mas seu conteúdo não pode ser visualizado.
Para maior segurança, ao usar um driver JDBC na configuração do pipeline, anexe allowLoadLocalInfile=false&autoDeserialize=false ao parâmetro jdbc_connection_string. Caso contrário, o sistema de agendamento reportará falha de validação ao adicionar o arquivo de configuração do Logstash. Exemplo: jdbc_connection_string => "jdbc:mysql://drdshbga51x6****.drds.aliyuncs.com:3306/<Your database name>?allowLoadLocalInfile=false&autoDeserialize=false".

Para mais detalhes sobre como configurar os parâmetros no campo Config Settings, consulte Logstash configuration files.

Clique em Next step e configure os parâmetros do pipeline.

Parâmetro	Descrição
Pipeline Workers	Número de threads de trabalho para as etapas de filtro e saída. Aumente este valor se houver acúmulo de eventos ou subutilização da CPU. Padrão: número de núcleos da CPU.
Pipeline Batch Size	Quantidade máxima de eventos que um worker coleta antes de executar filtros e saídas. Lotes maiores aumentam o uso de memória e podem exigir um heap JVM maior (LS_HEAP_SIZE). Padrão: 125.
Pipeline Batch Delay	Tempo de espera em milissegundos antes de enviar um lote incompleto para uma thread de trabalho. Padrão: 50 ms.
Queue Type	Modelo interno de fila para buffer de eventos entre etapas. Valores válidos: MEMORY: Padrão. Utiliza uma fila em memória. PERSISTED: Fila persistente baseada em disco.
Queue Max Bytes	Capacidade máxima de armazenamento da fila, em `MB`. O valor deve ser um inteiro entre `1` e `2⁵³-1`. Valor padrão: `1024`. Nota Certifique-se de que este valor seja menor que a capacidade total do disco.
Queue Checkpoint Writes	Número máximo de eventos gravados antes de forçar um checkpoint (apenas para fila persistente). 0 significa sem limite. Padrão: 1024.

Aviso

Salvar e implantar a configuração reinicia a instância. Prossiga apenas se isso não impactar sua aplicação.

Clique em Save ou Save and Deploy.
- Save: Salva a configuração do pipeline sem aplicá-la. Após salvar, você retorna à página Pipelines. Na seção Pipelines, clique em Deploy Now na coluna Actions para reiniciar a instância e aplicar a configuração.
- Save and Deploy: Salva e implanta a configuração, reiniciando a instância para aplicar as alterações.

Etapa 3: Verificar o resultado

Faça login no console Kibana do cluster do Elasticsearch.

Para mais informações, consulte Log on to the Kibana console.
No painel de navegação à esquerda, clique em Dev Tools.
No Console, execute o comando abaixo para verificar a quantidade de documentos sincronizados.
```
GET drds_test/_count
{
  "query": {"match_all": {}}
}
```
Se o comando for executado com sucesso, o seguinte resultado será retornado:
```
{
  "count" : 3,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  }
}
```

Atualize e insira dados na tabela.

UPDATE food SET name='Chocolates',update_time=now() where id = 1;
INSERT INTO food values(null,'egg',now(),now());

Visualize os dados atualizados e inseridos.

Consulte o registro cujo valor de name é Chocolates.

GET drds_test/_search
{
  "query": {
    "match": {
      "name": "Chocolates"
   }}
}

A consulta é bem-sucedida e retorna um documento correspondente. A resposta é:

{
  "took" : 18,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 1,
    "max_score" : 1.7427701,
    "hits" : [
      {
        "_index" : "drds_test",
        "_type" : "doc",
        "_id" : "1",
        "_score" : 1.7427701,
        "_source" : {
          "update_time" : "2020-08-05T09:02:31.000Z",
          "id" : 1,
          "insert_time" : "2020-08-05T04:00:01.000Z",
          "@timestamp" : "2020-08-05T09:03:00.355Z",
          "@version" : "1",
          "name" : "Chocolates"
        }
      }
    ]
  }
}

Consulte todos os dados.

GET drds_test/_search
{
  "query": {
    "match_all": {}
  }
}

Se o comando for executado com sucesso, o seguinte resultado será retornado:

{
  "hits" : {
    "hits" : [
      {
        "_index" : "drds_test",
        "_type" : "doc",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "update_time" : "2020-08-05T04:00:02.000Z",
          "id" : 3,
          "insert_time" : "2020-08-05T04:00:02.000Z",
          "@timestamp" : "2020-08-05T09:03:00.356Z",
          "@version" : "1",
          "name" : "火腿肠"
        }
      },
      {
        "_index" : "drds_test",
        "_type" : "doc",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "update_time" : "2020-08-05T04:00:02.000Z",
          "id" : 2,
          "insert_time" : "2020-08-05T04:00:02.000Z",
          "@timestamp" : "2020-08-05T09:03:00.355Z",
          "@version" : "1",
          "name" : "酸奶"
        }
      },
      {
        "_index" : "drds_test",
        "_type" : "doc",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "update_time" : "2020-08-05T09:02:32.000Z",
          "id" : 5,
          "insert_time" : "2020-08-05T09:02:32.000Z",
          "@timestamp" : "2020-08-05T09:08:00.310Z",
          "@version" : "1",
          "name" : "鸡蛋"
        }
      }
    ]
  }
}

Perguntas frequentes

FAQ about data transfer by using Logstash