Elasticsearch:Use Logstash to synchronize data from ApsaraDB RDS for MySQL to Elasticsearch

Limits

• Make sure that the ApsaraDB RDS for MySQL instance, the Alibaba Cloud Logstash cluster, and the Alibaba Cloud Elasticsearch cluster reside in the same time zone. Otherwise, when time-related data is synchronized, a time zone difference may exist between the data before and after synchronization.

• The _id field in Elasticsearch must be the same as the id field in MySQL.

  This condition ensures that when a record in MySQL is written to Elasticsearch, the synchronization task can establish a direct mapping between the MySQL record and the Elasticsearch document. For example, when a record is updated in MySQL, the synchronization task overwrites the Elasticsearch document that has the same ID as the updated record.

  Note

  Based on the internals of Elasticsearch, an update is essentially a deletion of the old document followed by the indexing of the new document. Therefore, overwriting a document in Elasticsearch is as efficient as an update operation.

• When data is inserted or updated in MySQL, the corresponding record must contain a field that stores the update time or the insert time.

  Each time Logstash polls MySQL, Logstash saves the update time or insert time of the last record that it read from MySQL. When Logstash reads data, Logstash reads only the records that meet the condition, that is, the records whose update time or insert time is later than that of the last record in the previous poll.

  Important

  The logstash-input-jdbc plug-in cannot synchronize deletions. You must run the relevant command in Elasticsearch to delete the documents manually.

Prerequisites

We recommend that you create the following instances in the same virtual private cloud (VPC):

You can also use services that are deployed on the Internet. In this case, you must configure SNAT, enable the public endpoint of the ApsaraDB RDS for MySQL instance, and remove the whitelist limit. For more information about how to configure SNAT, see Configure a NAT gateway for public data transfer. For more information about how to configure a whitelist, see Configure an IP address whitelist.

• Create an ApsaraDB RDS for MySQL instance. For more information, see Create an ApsaraDB RDS for MySQL instance. In this topic, MySQL 5.7 is used.

• Create an Alibaba Cloud Elasticsearch cluster. For more information, see Create an Alibaba Cloud Elasticsearch cluster. In this topic, an Elasticsearch V8.17 cluster is used.

• Create an Alibaba Cloud Logstash cluster. For more information, see Create an Alibaba Cloud Logstash instance. In this topic, a Logstash V8.11.4 cluster is used.

Background information

Alibaba Cloud Logstash is a data collection and processing tool that provides data collection, transformation, optimization, and output capabilities. The logstash-input-jdbc plug-in of Logstash is installed by default and cannot be uninstalled. You can use the plug-in to query data in an ApsaraDB RDS for MySQL instance in batches and synchronize the data to Elasticsearch. The plug-in also periodically polls the data in the ApsaraDB RDS for MySQL instance and synchronizes the records that were inserted or changed since the previous poll to Elasticsearch. For more information, see How to keep Elasticsearch synchronized with a relational database using Logstash in the official documentation. This solution is suitable for scenarios in which full data is synchronized and a latency of seconds is acceptable, or scenarios in which data that meets specific conditions is queried in batches and then synchronized.

Synchronize data

FAQ

My pipeline is stuck in the initializing state, data is inconsistent after sync, or the database connection fails. What should I do?

Check the Logstash cluster logs first. Go to the Logstash console and use the Query logs feature to view error details. For details, see Query logs.

If a cluster update is in progress when you apply a fix, pause the update first. See View the progress of a cluster task. After the fix, the system restarts the cluster and resumes the update automatically.

The following table lists common causes and solutions:

For more troubleshooting guidance, see FAQ about data transfer by using Logstash.

How do I sync data from multiple MySQL tables to separate Elasticsearch indexes?

Define multiple jdbc blocks in the input section, assign a type value to each, and use if[type] conditions in the output section to route each table's data to a different index:

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