added example for MySQL

docker-compose.yml

@@ -0,0 +1,31 @@
+version: '3.7'
+
+services:
+
+  mysql:
+    image: mysql:5.7
+    environment:
+      - MYSQL_ROOT_PASSWORD=s3cr3t
+      - MYSQL_DATABASE=blog
+      - MYSQL_USER=blog
+      - MYSQL_PASSWORD=blog
+    volumes:
+    - ./mysql/schema.sql:/docker-entrypoint-initdb.d/00_schema.sql
+    ports:
+    - 3306:3306
+
+  es:
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.5.2
+    environment:
+      - discovery.type=single-node
+    ports:
+    - 9200:9200
+    - 9300:9300
+
+  logstash:
+    image: docker.elastic.co/logstash/logstash:7.5.2
+    environment:
+      - xpack.monitoring.enabled=false
+    volumes:
+    - ./logstash:/usr/share/logstash/pipeline
+    - ./rdbc:/rdbc

logstash/sync-pipeline.conf

@@ -0,0 +1,29 @@
+input {
+  jdbc {
+    jdbc_driver_library => "/rdbc/mysql-connector-java-8.0.19.jar"
+    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
+    jdbc_connection_string => "jdbc:mysql://mysql:3306/blog"
+    jdbc_user => "blog"
+    jdbc_password => "blog"
+    jdbc_paging_enabled => true
+    tracking_column => "unix_ts_in_secs"
+    use_column_value => true
+    tracking_column_type => "numeric"
+    schedule => "*/5 * * * * *"
+    statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM articles WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC"
+  }
+}
+filter {
+  mutate {
+    copy => { "id" => "[@metadata][_id]"}
+    remove_field => ["id", "@version", "unix_ts_in_secs"]
+  }
+}
+output {
+  # stdout { codec =>  "rubydebug"}
+  elasticsearch {
+      hosts => ["es:9200"]
+      index => "articles"
+      document_id => "%{[@metadata][_id]}"
+  }
+}

mysql/schema.sql

@@ -0,0 +1,10 @@
+DROP TABLE IF EXISTS articles;
+CREATE TABLE articles
+(
+    id                VARCHAR(32)   NOT NULL,
+    title             VARCHAR(255)  NOT NULL,
+    content           VARCHAR(2000) NOT NULL,
+    modification_time TIMESTAMP     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    created_time      TIMESTAMP     NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (id)
+);