How to configure a Logstash JDBC input

A Logstash JDBC input keeps relational database tables searchable and analyzable without periodic full exports. Scheduled SQL polling makes it practical to ship application data into Elasticsearch for dashboards, alerting, and fast queries while keeping the database as the source of truth.

The jdbc input plugin loads a vendor JDBC driver jar, connects using a JDBC connection string, and runs a SQL statement on a schedule. Each returned row is converted into a Logstash event, so column names become fields that can be filtered and routed to outputs such as Elasticsearch.

Incremental imports rely on sql_last_value, stored in a metadata file, to resume from the last seen value instead of re-importing entire tables. The tracking column must be indexed and the statement must use a deterministic ORDER BY to prevent gaps or duplicates, especially when multiple rows share the same timestamp or when paging is enabled. Storing database credentials in plain text inside /etc/logstash/conf.d increases exposure risk, so secret handling should be planned before running in production.

Steps to configure a JDBC input in Logstash:

Copy the database JDBC driver jar to /usr/share/logstash/vendor/jdbc/mysql-connector-j.jar.
```
$ sudo install -D -m 0644 mysql-connector-j.jar /usr/share/logstash/vendor/jdbc/mysql-connector-j.jar
```
Update jdbc_driver_library and jdbc_driver_class to match the database vendor.
Create a state directory for sql_last_value under /var/lib/logstash/jdbc_last_run.
```
$ sudo install -o logstash -g logstash -m 0750 -d /var/lib/logstash/jdbc_last_run
```
Use a unique last_run_metadata_path per JDBC input to avoid sharing checkpoints across pipelines.

Create a pipeline configuration file at /etc/logstash/conf.d/30-jdbc.conf.

Plain-text passwords inside /etc/logstash/conf.d are readable by anyone with access to the file, so avoid committing credentials into version control or sharing backups broadly.

Using document_id makes repeated rows idempotent when updated_at has coarse resolution and the query uses >= :sql_last_value.

input {
  jdbc {
    id => "jdbc_events"
    jdbc_driver_library => "/usr/share/logstash/vendor/jdbc/mysql-connector-j.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://db.example.net:3306/app"
    jdbc_user => "logstash"
    jdbc_password => "logstash-db-pass"
    schedule => "*/5 * * * *"
    statement => "SELECT id, event_type, payload, updated_at FROM events WHERE updated_at >= :sql_last_value ORDER BY updated_at ASC, id ASC"
    use_column_value => true
    tracking_column => "updated_at"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/var/lib/logstash/jdbc_last_run/events.yml"
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch.example.net:9200"]
    index => "db-events-%{+YYYY.MM.dd}"
    document_id => "%{id}"
  }
}

Test the pipeline configuration.

$ sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash --config.test_and_exit
Using bundled JDK: /usr/share/logstash/jdk
Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties
[2026-01-08T08:34:11,845][WARN ][logstash.runner          ] NOTICE: Running Logstash as a superuser is strongly discouraged as it poses a security risk. Set 'allow_superuser' to false for better security.
##### snipped #####
Configuration OK
[2026-01-08T08:34:12,390][INFO ][logstash.runner          ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

Restart the Logstash service to load the JDBC input.
```
$ sudo systemctl restart logstash
```
Related: How to manage the Logstash service with systemctl in Linux

Confirm the JDBC input is running in the main pipeline.

$ curl -s http://localhost:9600/_node/stats/pipelines/main?pretty
{
  "pipelines" : {
    "main" : {
      "events" : {
        "filtered" : 3,
        "in" : 3,
        "out" : 3,
        "duration_in_millis" : 476,
        "queue_push_duration_in_millis" : 0
      },
      "plugins" : {
        "inputs" : [ {
          "id" : "jdbc_events",
          "name" : "jdbc",
          "events" : {
            "out" : 3,
            "queue_push_duration_in_millis" : 0
          }
        } ]
      }
    }
  }
##### snipped #####
}

Confirm the sql_last_value checkpoint file exists after the first successful run.

$ sudo cat /var/lib/logstash/jdbc_last_run/events.yml
--- 2026-01-08 08:22:00.000000000 Z