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

How it works

The plug-in queries ApsaraDB RDS for MySQL on a configurable schedule using a SQL statement. On each run, it records the highest value of a tracking column (such as update_time) and uses that value as the lower bound for the next query. This polling approach captures newly inserted and updated rows, but it cannot detect deleted rows — deletions must be handled separately in Elasticsearch.

Limitations

Understand these constraints before you begin:

• Same time zone required. The RDS instance, Elasticsearch cluster, and Logstash cluster must be in the same time zone. A mismatch causes timestamp fields to have a time zone offset after synchronization.

• Document IDs must map to the MySQL primary key. Set document_id to %{id} in the output configuration, where id is the primary key column in MySQL. This mapping allows the plug-in to overwrite the correct Elasticsearch document when a row is updated in MySQL. Elasticsearch handles overwrites the same way as a standard update operation: it deletes the old document and indexes the new one.

• Deletions are not propagated. The plug-in queries only rows inserted or updated after sql_last_value, so it cannot detect DELETE operations in MySQL. To remove a document from Elasticsearch, run the corresponding delete command directly on the cluster.

• Each row must include an insert/update timestamp column. The plug-in uses this column as the tracking column to determine which rows to fetch on the next run. By default, sql_last_value starts at Thursday, 1 January 1970.

• Security parameters are mandatory. Always append allowLoadLocalInfile=false&autoDeserialize=false to jdbc_connection_string. Without these parameters, the pipeline configuration check fails.

Prerequisites

Before you begin, ensure that you have:

• An ApsaraDB RDS for MySQL instance (this example uses MySQL 5.7)

• An Alibaba Cloud Elasticsearch cluster (this example uses Elasticsearch V7.10)

• An Alibaba Cloud Logstash cluster

Deploy all three resources in the same virtual private cloud (VPC) to keep traffic private and avoid NAT configuration. If your RDS instance uses a public endpoint, configure a Source Network Address Translation (SNAT) entry for the Logstash cluster, enable the public endpoint on the RDS instance, and add the Logstash node IP addresses to the RDS whitelist. For details, see Configure a NAT gateway for data transmission over the Internet.

Synchronize data from ApsaraDB RDS for MySQL to Elasticsearch

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