Elasticsearch:Use Logstash to synchronize data from PolarDB-X (DRDS) to Elasticsearch

Create a table in PolarDB-X 1.0 and insert test data. The following example creates a food table:

CREATE TABLE food(
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(32),
  insert_time DATETIME,
  update_time DATETIME
);

Key columns:

• id: The primary key. Its value is mapped to the _id field in Elasticsearch. Required for the sync pipeline to correctly identify and overwrite documents.

• update_time: The tracking timestamp. The logstash-input-jdbc plug-in uses this field to detect changes since the last poll. Required for incremental sync to work.

• insert_time: Records when the row was first created. Not required for sync, but useful for auditing.

Insert sample data:

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());

Enable the Auto Indexing feature for your Elasticsearch cluster. For details, see Access and configure an Elasticsearch cluster.

Upload a MySQL JDBC driver to the Logstash cluster. The driver version must be compatible with your PolarDB-X 1.0 instance. This example uses mysql-connector-java-5.1.35. For upload instructions, see Configure third-party libraries.

Add the IP addresses of the Logstash cluster nodes to the IP address whitelist of your PolarDB-X 1.0 instance. Find the node IP addresses on the Basic Information page of the Logstash cluster in the Elasticsearch console. For whitelisting instructions, see Set an IP address whitelist.

Go to the Logstash Clusters page of the Alibaba Cloud Elasticsearch console.

In the top navigation bar, select the region where your cluster resides.

On the Logstash Clusters page, find your cluster and click its ID.

In the left-side navigation pane, click Pipelines.

On the Pipelines page, click Create Pipeline.

On the Create Task page, enter a pipeline ID in the Pipeline ID field, then enter the following configuration in the Config Settings field. Replace <Logstash cluster ID>, <Database name>, and the credentials with your actual values. To find your Logstash cluster ID, see Overview of the Logstash Clusters page.

Input plug-in parameters

Output plug-in parameters

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

For the full list of input plug-in parameters, see Logstash JDBC input plugin. For general pipeline configuration syntax, see Logstash configuration files.

Click Next to configure pipeline parameters.

Click Save or Save and Deploy.

• Save: Stores the pipeline configuration and triggers a cluster change, but does not activate the pipeline. To activate it, go to the Pipelines page, find the pipeline, and click Deploy in the Actions column.

• Save and Deploy: Saves and immediately restarts the Logstash cluster to activate the pipeline.

Log on to the Kibana console of the Elasticsearch cluster. For details, see Log on to the Kibana console.

In the left-side navigation pane, click Dev Tools.

On the Console tab, run the following query to check that the three sample records were synced:

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

Expected response:

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

Update an existing record and insert a new one in PolarDB-X 1.0:

UPDATE food SET name='Chocolates', update_time=NOW() WHERE id = 1;
INSERT INTO food VALUES(null, 'Egg', NOW(), NOW());

After the next polling cycle completes (within one minute), verify the updates in Kibana. Query the updated record:

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

If the command runs successfully, the result shown in the following figure is returned. Query all records to confirm the new row was added:

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

If the command runs successfully, the result shown in the following figure is returned.