1、前置准备

创建mysql.conf

input {
    stdin{
    }
    jdbc {
        # mysql 数据库连接
        jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/es?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC"
        jdbc_user => "root"
        jdbc_password => "root"
        # 驱动配置
        jdbc_driver_library => "/usr/local/data/logstash/driver/mysql-connector-java-8.0.19.jar"
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        # 执行的sql文件
        statement_filepath => "/usr/local/data/logstash/mysqlconfig/blog.sql"
         # 设置监听 各字段含义 分 时 天 月  年 ，默认全部为*代表含义：每分钟都更新
        schedule => "* * * * *"
        # 索引类型
        type => "blog"
    }
}
 
 
output {
    elasticsearch {
        #es服务器
        hosts => ["localhost:9200"]
        # es索引名称
        index => "sl_blog"
        #id自增长
        document_id => "%{id}"
    }
    
    stdout {
        codec => json_lines
    }
}

创建blog.sql

select * from t_blog

将mysql驱动包添加到指定文件夹

/usr/local/data/logstash/driver/mysql-connector-java-8.0.19.jar

创建测试数据库

# 创建数据库
create database es;
use es;
# 创建表
CREATE TABLE `t_blog` (
  `id` varchar(32) CHARACTER SET utf8 COLLATE utf8_unicode_ci NOT NULL,
  `title` varchar(32) DEFAULT NULL,
  `create_time` datetime DEFAULT NULL,
  `content` varchar(2000) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
# 添加数据
INSERT INTO `t_blog` VALUES ('1', '中国', '2019-04-24 15:22:30', '中国当自强');
INSERT INTO `t_blog` VALUES ('2', 'myblog', '2019-04-24 15:22:50', 'my blog content');
INSERT INTO `t_blog` VALUES ('3', '上海', '2019-04-24 15:23:08', '中国上海');
INSERT INTO `t_blog` VALUES ('4', '成都', '2020-01-03 01:48:15', '四川省会');

2、执行

logstash -f ../mysql/mysql.conf

3、访问kinaba验证es数据