Integrating MySQL with Elasticsearch via Logstash: Building an Automated Data Sync Pipeline – ITFROMZERO

Table of Contents

Real-World Problem: MySQL Overload as Data Grows

When the products table hits the 500,000-row mark, search performance starts degrading. Users often search with unaccented keywords, typos, or apply 4–5 filters simultaneously: category, price range, ratings, and inventory.

If you keep using MySQL with LIKE '%keyword%' and 3–4 JOIN clauses, database CPU utilization quickly hits 100%. Response times can spike from 50ms to 3–5 seconds. As a result, the entire connection pool gets exhausted, bringing order creation and checkout APIs to a halt.

Why Does MySQL Struggle with Full-Text Search?

There are 3 technical reasons why MySQL is ill-suited for advanced search workloads:

B-Tree Indexes Do Not Support Leading Wildcards: MySQL is strictly optimized for exact matches or prefix searches (LIKE 'iphone%'). When you place a % at the beginning (LIKE '%iphone%'), the index becomes completely useless, forcing MySQL to perform a Full Table Scan.
Read-Write Resource Contention: MySQL is designed for OLTP workloads (ACID transactions). Forcing the database to compute relevance scoring while concurrently handling transactional writes increases lock contention and pushes disk I/O to critical levels.
Elasticsearch’s Inverted Index Architecture: Elasticsearch tokenizes text into discrete terms and builds an inverted index. During searches, it simply queries the term dictionary to retrieve the matching document IDs. Response times typically stay under 30ms, even across datasets with tens of millions of records.

3 Common Data Synchronization Approaches

Depending on your system’s scale, you can choose one of the following approaches:

Application-Level Dual-Write: Whenever a record is created or updated in MySQL, the backend service publishes an event via RabbitMQ or Kafka to update Elasticsearch. This approach provides near-instant updates. However, it adds boilerplate code and easily causes data discrepancies during network failures.
Change Data Capture (CDC) with Debezium: A solution that reads directly from the MySQL binlog. It is highly accurate and provides true real-time synchronization. In return, it comes with high infrastructure and operational complexity, making it worthwhile primarily for large microservice architectures.
Using Logstash with JDBC Input Plugin: Logstash acts as an intermediate worker, periodically querying new or updated records from MySQL and pushing them to Elasticsearch.

The Pragmatic Solution: Automated Pipeline with Logstash JDBC

For most small to medium-sized applications (under a few million records), Logstash JDBC is the most balanced choice. It requires zero backend code changes, provides centralized configuration, and takes only about 30 minutes to set up.

Here is a practical 4-step setup guide.

Step 1: Prepare the MySQL Database Table

For Logstash to identify which records need syncing without scanning millions of rows every time, the table must include an indexed updated_at column.

CREATE DATABASE IF NOT EXISTS shop_db;
USE shop_db;

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    description TEXT,
    price DECIMAL(10, 2) NOT NULL,
    is_deleted TINYINT(1) DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_updated_at (updated_at)
);

-- Insert sample data
INSERT INTO products (title, description, price) VALUES
('Wireless Mechanical Keyboard', '75% layout, Bluetooth and 2.4GHz connectivity', 1500000),
('Ergonomic Gaming Mouse', '60g weight, 26000 DPI sensor', 850000),
('27-inch 4K IPS Monitor', '100% sRGB color gamut, designed for graphic design', 7200000);

Step 2: Download MySQL Connector/J (JDBC Driver)

Logstash requires a JDBC driver to communicate with MySQL. Download the .jar file to your Logstash server:

# Create directory for the driver
sudo mkdir -p /etc/logstash/drivers
cd /etc/logstash/drivers

# Download MySQL Connector/J (version 8.0.33)
sudo wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.33/mysql-connector-java-8.0.33.jar

Step 3: Configure the Logstash Pipeline

Create a configuration file at /etc/logstash/conf.d/mysql_to_es.conf:

input {
  jdbc {
    jdbc_driver_library => "/etc/logstash/drivers/mysql-connector-java-8.0.33.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/shop_db?useSSL=false&serverTimezone=UTC"
    jdbc_user => "db_user"
    jdbc_password => "Secret_Password_123"
    
    # Run interval: every 1 minute
    schedule => "* * * * *"
    
    # Track update timestamp
    use_column_value => true
    tracking_column => "updated_at"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/var/lib/logstash/.logstash_products_last_run"
    
    # Fetch only records newer than the last run timestamp
    statement => "SELECT id, title, description, price, is_deleted, updated_at FROM products WHERE updated_at > :sql_last_value ORDER BY updated_at ASC"
  }
}

filter {
  mutate {
    remove_field => ["@version", "@timestamp"]
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "products"
    document_id => "%{id}"
    action => "index"
  }
  
  # Output logs to console for debugging if needed
  stdout {
    codec => rubydebug
  }
}

Step 4: Start and Verify the Pipeline

First, verify the validity of your configuration file:

# Validate configuration syntax
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/mysql_to_es.conf --config.test_and_exit

# Start and enable the Logstash service
sudo systemctl start logstash
sudo systemctl enable logstash

After about a minute, verify that data has landed in Elasticsearch using cURL:

curl -X GET "http://localhost:9200/products/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "query": {
    "match": {
      "title": "mechanical keyboard"
    }
  }
}'

Production Best Practices

A pipeline running smoothly in testing doesn’t guarantee smooth operation in production. Keep these 3 critical considerations in mind:

Handling Deletions (Hard Delete vs. Soft Delete): Because Logstash executes periodic SELECT queries, it cannot detect hard DELETE operations. Therefore, adopt a Soft Delete strategy (flagging is_deleted = 1). In Elasticsearch, simply filter out documents with this flag when querying.
Timezone Synchronization: Ensure UTC is consistently configured across MySQL, the Linux host, and Logstash. The serverTimezone=UTC parameter in the JDBC connection string is required to prevent timezone shifts from causing :sql_last_value to miss data.
Keep MySQL as the Single Source of Truth: Elasticsearch should only serve as a search index/cache. Never read directly from Elasticsearch for transactional logic like payments or inventory deductions. Maintain regular MySQL backups and clearly define primary-to-secondary data flows.

Combining MySQL with Elasticsearch via Logstash allows you to fully harness Elasticsearch’s superior search performance while keeping your application architecture clean, consistent, and maintainable.