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

Background information

Alibaba Cloud Logstash is a powerful data processing tool that you can use to collect, transform, optimize, and generate data. Alibaba Cloud Logstash provides the logstash-input-jdbc plug-in, which is installed on Logstash clusters by default and cannot be removed. The plug-in can query multiple data records in ApsaraDB RDS for MySQL at a time and synchronize the data records to Alibaba Cloud Elasticsearch. The plug-in uses a round-robin method to identify the most recently inserted or updated data records in ApsaraDB RDS for MySQL on a regular basis and synchronizes the data records to Alibaba Cloud Elasticsearch. For more information, see Use Logstash and JDBC to synchronize data from a relational database to an Elasticsearch cluster. If you want to synchronize full data and can accept latency of a few seconds, or you want to query specific data at a time and synchronize the data, you can use Alibaba Cloud Logstash.

Prerequisites

We recommend that you create the following instance and clusters in the same virtual private cloud (VPC).

• Create an ApsaraDB RDS for MySQL instance. For more information, see Create an ApsaraDB RDS for MySQL instance. In this example, an ApsaraDB RDS instance that runs MySQL 5.7 is created.

• Create an Alibaba Cloud Elasticsearch cluster. For more information, see Create an Alibaba Cloud Elasticsearch cluster. In this example, an Elasticsearch V7.10 cluster is created.

• Create an Alibaba Cloud Logstash cluster. For more information, see Create an Alibaba Cloud Logstash cluster.

Limits

Procedure

FAQ

• Q: What do I do if my data synchronization task fails because the pipeline is stuck in the initializing state, data before and after synchronization is inconsistent, or the connection to the database fails?

  A: Check whether the cluster logs of your Logstash cluster contain error information and identify the cause based on the error information. For more information, see Query logs. The following table describes common causes of errors and solutions to the errors.

  Note

  If an update operation is being performed on your Logstash cluster when you perform the operations described in the following solutions, pause the update operation by referring to View the progress of a cluster task. After the operations described in the solutions are complete, the system restarts the Logstash cluster and resumes the update operation.

  Cause	Solution
  The IP addresses of nodes in the Logstash cluster are not added to a whitelist of the ApsaraDB RDS for MySQL instance.	Add the IP addresses of nodes in the Logstash cluster to a whitelist of the ApsaraDB RDS for MySQL instance. For more information, see Use a database client or the CLI to connect to an ApsaraDB RDS for MySQL instance. Note For more information about how to obtain the IP addresses of the nodes in a Logstash cluster, see View the basic information of a cluster.
  You use the Logstash cluster to synchronize data from a self-managed MySQL database that is hosted on an ECS instance, but the private IP addresses and internal ports of nodes in the cluster are not added to a security group of the ECS instance.	Add the private IP addresses and internal ports of nodes in the cluster to a security group of the ECS instance. For more information, see Add a security group rule. Note For more information about how to obtain the IP address and port of a node in a Logstash cluster, see View the basic information of a cluster.
  The Elasticsearch cluster does not reside in the same VPC as the Logstash cluster.	Use one of the following solutions: Purchase an Elasticsearch cluster that resides in the same VPC as the Logstash cluster. For more information, see Create an Alibaba Cloud Elasticsearch cluster. After the Elasticsearch cluster is purchased, modify the pipeline configuration of the Logstash cluster. Configure a NAT gateway to transmit data over the Internet. For more information, see Configure a NAT gateway for data transmission over the Internet.
  The endpoint of the ApsaraDB RDS for MySQL instance is incorrect, and the port number of the instance is not 3306.	Obtain the correct endpoint and port number. For more information, see View and manage instance endpoints and ports. Then, replace the endpoint and port number in the value of the jdbc_connection_string parameter with the endpoint and port number that you obtained. Important <Endpoint of the ApsaraDB RDS for MySQL instance>: You must specify the internal endpoint of the ApsaraDB RDS for MySQL instance. If you want to use the public endpoint of the ApsaraDB RDS for MySQL instance, you must configure a NAT gateway for the Logstash cluster to enable the Logstash cluster to connect to the Internet. For more information, see Configure a NAT gateway for data transmission over the Internet.
  The Auto Indexing feature is disabled for the Elasticsearch cluster.	Enable the Auto Indexing feature for the Elasticsearch cluster. For more information, see Configure the YML file.
  The load of the Elasticsearch or Logstash cluster is excessively high.	Upgrade the configuration of the Elasticsearch or Logstash cluster. For more information, see Upgrade the configuration of a cluster. Note If the load of the Elasticsearch cluster is excessively high, you can view the monitoring data collected based on metrics in the Elasticsearch console to obtain the load information of the Elasticsearch cluster. For more information, see Metrics and exception handling suggestions. If the load of the Logstash cluster is excessively high, you can enable the X-Pack Monitoring feature for the Logstash cluster, use the feature to monitor the Logstash cluster, and then view the monitoring data. For more information, see Enable the X-Pack Monitoring feature.
  No driver file that is used for JDBC-based connection to the ApsaraDB RDS for MySQL database is uploaded.	Upload a driver file. For more information, see Configure third-party libraries.
  file_extend is specified in the configuration of the pipeline. However, the logstash-output-file_extend plug-in is not installed.	Use one of the following solutions: Install the logstash-output-file_extend plug-in. For more information, see Install and remove a plug-in. Remove the file_extend parameter from the configuration of the pipeline.

  For more information about the causes of and solutions to this issue, see FAQ about data transfer by using Logstash.

• Q: How do I configure multiple JDBC connections to the source database in the input configuration of a pipeline?

  A: You can define multiple JDBC data sources in the input configuration of the pipeline and specify a query statement for each table in the statement parameter. The following code provides an example:

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

  In the preceding example, the type parameter is added to the jdbc configuration. This parameter helps you define conditions in the output configuration and synchronize data from different tables to different indexes.