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

Etapa 1: Preparar o ambiente

1. Ative o recurso Auto Indexing no cluster Elasticsearch para que o Logstash possa criar índices automaticamente. Para mais detalhes, consulte Acessar e configurar um cluster Elasticsearch.

2. Faça upload de um driver JDBC compatível com sua versão do MySQL para o cluster Logstash. Este exemplo usa mysql-connector-java-5.1.48.jar. Para mais detalhes, consulte Configurar bibliotecas de terceiros.

3. Crie uma tabela de teste e insira dados de amostra na instância RDS:

   CREATE table food (
     id int PRIMARY key AUTO_INCREMENT,
     name VARCHAR (32),
     insert_time DATETIME,
     update_time DATETIME
   );
   
   INSERT INTO food values(null,'Chocolates',now(),now());
   INSERT INTO food values(null,'Yogurt',now(),now());
   INSERT INTO food values(null,'Ham sausage',now(),now());

4. Adicione os endereços IP dos nós do cluster Logstash à lista de permissões do RDS. Encontre os endereços IP na página Basic Information do cluster Logstash.

Etapa 2: Configurar um pipeline do Logstash

1. Acesse a página Logstash Clusters no console Alibaba Cloud Elasticsearch.

2. Na barra de navegação superior, selecione a região onde o cluster está localizado. Na página Logstash Clusters, clique em Pipelines e, em seguida, clique em Create Pipeline.

3. No painel de navegação à esquerda, clique em Pipelines e, em seguida, clique em Create Pipeline.

4. Insira um Pipeline ID e cole a seguinte configuração no campo Config Settings. Substitua os valores de espaço reservado pelos seus próprios dados.

   Para obter o ID do cluster Logstash, consulte Visualizar as informações básicas de um cluster . Para a lista completa de parâmetros de input suportados, consulte Plug-in de input Jdbc do Logstash . Para parâmetros de output, consulte Arquivos de configuração do Logstash .

   input {
     jdbc {
       # JDBC driver class for MySQL
       jdbc_driver_class => "com.mysql.jdbc.Driver"
   
       # Path to the driver JAR uploaded in Step 1.2.
       # Format: /ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/<driver filename>
       jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
   
       # RDS connection string. Use the internal endpoint when instances share a VPC.
       # allowLoadLocalInfile=false and autoDeserialize=false are required — omitting them causes a check failure.
       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"
   
       # Credentials for the RDS database user
       jdbc_user => "xxxxx"
       jdbc_password => "xxxx"
   
       # Enable paging and set page size to control memory usage on large tables (default: false)
       jdbc_paging_enabled => "true"
       jdbc_page_size => "50000"
   
       # Fetch only rows updated after the last recorded value (incremental sync)
       statement => "select * from food where update_time >= :sql_last_value"
   
       # Run every minute (Rufus cron expression)
       schedule => "* * * * *"
   
       # Persist the last run value so the next run starts where this one ended
       record_last_run => true
   
       # Store the tracking value in this file. Use the /ssd/1/<cluster ID>/logstash/data/ path
       # to ensure write permissions. Replace <Logstash cluster ID> with your cluster ID.
       last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
   
       # Set clean_run to true to reset and re-sync all data from the beginning
       clean_run => false
   
       # Track by timestamp column to capture inserts and updates.
       # The values in this column must be sorted in ascending order.
       tracking_column_type => "timestamp"
       use_column_value => true
       tracking_column => "update_time"
     }
   }
   filter {
   }
   output {
     elasticsearch {
       hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
       index => "rds_es_dxhtest_datetime"
       user => "elastic"
       password => "xxxxxxx"
       # Map the Elasticsearch document ID to the MySQL primary key
       document_id => "%{id}"
     }
   }

5. Clique em Next para configurar os parâmetros do pipeline.

   Aviso

   Salvar e implantar o pipeline aciona uma reinicialização do cluster Logstash. Confirme se a reinicialização não afetará suas cargas de trabalho antes de prosseguir.

   Parâmetro	Descrição	Padrão
   Pipeline Workers	Quantidade de threads que executam plug-ins de filtro e output em paralelo. Aumente este valor quando os recursos de CPU estiverem subutilizados ou houver acúmulo de eventos.	Número de vCPUs
   Pipeline Batch Size	Máximo de eventos que um único worker coleta das entradas antes de executar filtros e outputs. Valores maiores aumentam o throughput, mas exigem mais memória heap da JVM.	125
   Pipeline Batch Delay	Tempo de espera do worker por eventos adicionais antes de iniciar um pequeno lote, em milissegundos.	50
   Queue Type	Modelo de fila interna para buffer de eventos. MEMORY: fila em memória. PERSISTED: fila baseada em disco com ACK para maior durabilidade.	MEMORY
   Queue Max Bytes	Tamanho máximo da fila no disco. Deve ser menor que a capacidade de disco disponível.	1024 MB
   Queue Checkpoint Writes	Máximo de eventos gravados antes que um checkpoint seja forçado (apenas filas persistentes). Defina como 0 para não haver limite.	1024

   

6. Clique em Save and Deploy para reiniciar o cluster Logstash e aplicar a configuração imediatamente. Alternativamente, clique em Save para armazenar a configuração e acionar uma alteração no cluster, mas as configurações ainda não entrarão em vigor. Para implantar posteriormente, acesse a página Pipelines, localize o pipeline e clique em Deploy Now na coluna Actions.

Etapa 3: Verificar a sincronização

1. Faça login no console Kibana do cluster Elasticsearch. Para mais detalhes, consulte Fazer login no console Kibana.

2. No canto superior esquerdo, clique em e escolha Management > Dev Tools.

3. Na aba Console, execute o seguinte comando para confirmar que as três linhas foram sincronizadas:

   GET rds_es_dxhtest_datetime/_count
   {
     "query": {"match_all": {}}
   }

   Resposta esperada:

   {
     "count" : 3,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     }
   }

4. Atualize e insira linhas no MySQL para testar a sincronização incremental:

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

5. Após a próxima execução agendada (até um minuto), consulte o Elasticsearch para confirmar as alterações:

   • Pesquise a linha atualizada:

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

     

   • Consulte todos os documentos:

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

     

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}"
        }
    }
}