logstash中介軟體實現全量，增量同步

1、下載安裝

選擇和ElasticSearch匹配的版本下載

2.操作

2.1下載合適jdbc jar包

這裡選擇mysql測試

2.2建立資料庫

CREATE DATABASE blog;

USE blog;

CREATE TABLE `t_blog` (    
   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',    
   `title` varchar(60) DEFAULT NULL COMMENT '部落格標題',    
   `author` varchar(60) DEFAULT NULL COMMENT '部落格作者',    
   `content` mediumtext COMMENT '部落格內容',    
   `create_time` datetime DEFAULT NULL COMMENT '建立時間',    
   `update_time` datetime DEFAULT NULL COMMENT '更新時間',    
   PRIMARY KEY (`id`)    
) ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4
 

2.3建立配置檔案

config下建立mysql.conf配置檔案

input{
    jdbc{
        # jdbc驅動包位置
        jdbc_driver_library => "E:\\tools\\logstash-7.10.1\\mysql-connector-java-8.0.22.jar"
        # 要使用的驅動包類
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        # mysql資料庫的連線資訊
        jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/blog"
        # mysql使用者
        jdbc_user => "root"
        # mysql密碼
        jdbc_password => "root"
        # 定時任務，多久執行一次查詢，預設一分鐘，如果想要沒有延遲，可以使用 schedule => "* * * * * *"
        schedule => "* * * * *"
        # 清空上傳的sql_last_value記錄
        clean_run => true
        # 你要執行的語句
        statement => "select * FROM t_blog WHERE update_time > :sql_last_value AND update_time < NOW() ORDER BY update_time desc"
    }
}

output {
    elasticsearch{
        # es host : port
        hosts => ["127.0.0.1:9200"]
        # 索引
        index => "blog"
        # _id
        document_id => "%{id}"
    }
}
 

2.4執行同步

首先確定ElasticSearch中沒有同步的資料

bin目錄下執行

logstash -f ../config/mysql.conf

配置是1分鐘執行一次

可以看出logstash確實在按我們配置的1分鐘同步一次，沒毛病。