Environment Setup

Download Logstash version matching your Elasticsearch installation (example uses 7.17.6):

• Logstash: Official Download Page
• MySQL Connector/J: Available from Maven repository or MySQL official site

Configure system environment variable LS_JAVA_HOME pointing to JDK 8 or 11. For Logstash 7.x, this variable is required instead of the traditional JAVA_HOME.

Directory Structure

Create a directory named mysql-config at the same level as the bin folder in your Logstash installation:

logstash-7.17.6/
├── bin/
├── mysql-config/
│   ├── database_sync.conf
│   ├── query_template.sql
│   ├── sync_tracker.dat
│   └── mysql-connector-java-8.0.xx.jar

Configuration Files

Main Configuration File

Create database_sync.conf with UTF-8 encoding (avoid BOM):

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://localhost:3306/sample_db?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC&useSSL=false"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_driver_library => "./mysql-config/mysql-connector-java-8.0.xx.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_paging_enabled => true
    jdbc_page_size => 5000
    
    statement_filepath => "./mysql-config/query_template.sql"
    schedule => "*/10 * * * *"
    lowercase_column_names => false
    codec => plain { charset => "UTF-8" }
    
    use_column_value => true
    tracking_column => "modified_at"
    tracking_column_type => "timestamp"
    record_last_run => true
    last_run_metadata_path => "./mysql-config/sync_tracker.dat"
    clean_run => false
    type => "students"
  }
  
  jdbc {
    jdbc_connection_string => "jdbc:mysql://localhost:3306/sample_db?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC&useSSL=false"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_driver_library => "./mysql-config/mysql-connector-java-8.0.xx.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_paging_enabled => true
    jdbc_page_size => 5000
    
    statement_filepath => "./mysql-config/staff_query.sql"
    schedule => "*/15 * * * *"
    lowercase_column_names => false
    codec => plain { charset => "UTF-8" }
    
    use_column_value => true
    tracking_column => "updated_timestamp"
    tracking_column_type => "timestamp"
    record_last_run => true
    last_run_metadata_path => "./mysql-config/staff_tracker.dat"
    clean_run => false
    type => "staff"
  }
}

filter {
  # Add data transformation logic here if needed
}

output {
  if [type] == "students" {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "elastic"
      password => "elastic_password"
      index => "student_records"
      document_id => "%{record_id}"
    }
  }
  
  if [type] == "staff" {
    elasticsearch {
      hosts => ["localhost:9200"]
      user => "elastic"
      password => "elastic_password"
      index => "employee_data"
      document_id => "%{emp_id}"
    }
  }
  
  stdout { codec => json_lines }
}

SQL Query Template

Create query_template.sql:

SELECT 
    id AS record_id,
    full_name,
    age_group,
    contact_number,
    modified_at
FROM students 
WHERE modified_at > DATE_ADD(:sql_last_value, INTERVAL 8 HOUR)
ORDER BY modified_at ASC

Synchronization Tracker

Create sync_tracker.dat with initial timestamp:

1970-01-01 00:00:00

This file automatically updates with the latest processed timestamp during execution.

Elasticsearch Index Mapping

Create indices in Kibana Dev Tools:

PUT /student_records
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "record_id": {"type": "long"},
      "full_name": {"type": "text"},
      "age_group": {"type": "keyword"},
      "contact_number": {"type": "keyword"},
      "modified_at": {"type": "date"}
    }
  }
}

Scheduling Syntax

The schedule parameter follows cron-like syntax:

* * * * *
│ │ │ │ │
│ │ │ │ └── Day of week (0-7, where 0 and 7 = Sunday)
│ │ │ └──── Month (1-12)
│ │ └────── Day of month (1-31)
│ └──────── Hour (0-23)
└────────── Minute (0-59)

Examples:

# Every 5 minutes
*/5 * * * *

# Every hour at minute 30
30 * * * *

# Daily at midnight
0 0 * * *

# First day of every month at 3 AM
0 3 1 * *

Execution Methods

Direct Execution

cd bin/
logstash -f ../mysql-config/database_sync.conf

Windows Service Installation

1. Install NSSM (Non-Sucking Service Manager)
2. Run nssm install from Logstash bin directory
3. Configure service parameters:
   • Path: Select logstash.bat
   • Arguments: -f ../mysql-config/database_sync.conf
   • Service name: Define custom name
4. Start the Windows service