使用logstash 同步数据库数据到elasticsearch

1.下载logstash 并解压
如图所示

测试logstash 的命令 这里就不提示了 网上有 主要是 logstash -e …
但是实际业务环境中 使用 logstash -f filepath
2.进入logstash-5.6.8 在目录下新建一个mysqletc文件夹
在mysqletc文件夹下导入mysql-connector-java的jar包 并创建mysql.conf 内容如下

input {
  jdbc {
	  # mysql jdbc connection string to our backup databse
	  #数据库地址
	  jdbc_connection_string => "jdbc:mysql://127.0.0.1:3307/tensquare_article?characterEncoding=utf-8"
	  # the user we wish to excute our statement as
	  #用户名
	  jdbc_user => "root"
	  #密码
	  jdbc_password => "123456"
	  # the path to our downloaded jdbc driver  
	  #数据库驱动路径
	  jdbc_driver_library => "C:\Users\Administrator\Desktop\tensqure\logstash-5.6.8\logstash-5.6.8\mysqletc\mysql-connector-java-5.1.46.jar"
	  # the name of the driver class for mysql
	  jdbc_driver_class => "com.mysql.jdbc.Driver"
	  #是否分页
	  jdbc_paging_enabled => "true"
	  jdbc_page_size => "50"
	  #以下对应着要执行的sql的绝对路径。
	  #statement_filepath => ""
	  #sql语句
	  statement => "SELECT id,title,content,state FROM tb_article"
	  #定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新（测试结果，不同的话请留言指出）
      schedule => "* * * * *" 
      #一分钟同步一次
  }
}

output {
  elasticsearch {
	  #ESIP地址与端口
	  hosts => "127.0.0.1" 
	  #ES索引名称（自己定义的）
	  index => "tensquare_article"
	  #自增ID编号
	  document_id => "%{id}" # 与sql语句查出来的id一样
	  document_type => "article"
  }
  stdout {
      #以JSON格式输出
      codec => json_lines
  }
}

3.进入logstash bin 目录 cmd进入命令行

执行 logstash -f ../mysqletc/mysql.conf

这里 …/是相对路径 往bin目录上级目录 再找到上级目录下的mysqletc(就是我们刚才创建的文件夹)找到mysql.conf 配置文件 或文件夹名字都可以随便起

观察到如下结果 则将数据库数据同步到了elasticsearch

我们访问 localhost:9100 进入head插件 查看是否同步成功

显示已同步成功，且是最小单位 一分钟同步一次

4.遇到的问题

提示

error:reason=>"Expected one of #, input, filter, output at line 1,
column 1 (byte 1) after "}

网上的解决方法：

Windows文件默认是utf8 。将文件格式改为：ansi格式就可以了。程序可以正常运行 bin\logstash -f
./config/test.conf

在这里 我试了一下 重新另存为ansi格式还是会报如下错误：

config files contains non-ascii characters but are not UTF-8 encoded

最后我找到了解决办法：
用editplus 打开mysql.conf文件 另存为格式必须为UTF-8 而不是UTF-8BOM 再次运行就成功了！***,

注意 在使用logstash同步过程中，一般我们要将数据库某个数据删除 ，都采用逻辑删除 ，更改状态，再同步到elasticsearch中。如果是直接delete语句 那么将不会同步到elasticsearch中，要逻辑删除，才能同步到elasticsearch.