logstash把mysql数据导入elasticsearch

admin

下载logstash: https://www.elastic.co/fr/downloads/logstash
解压到自定义目录，我的是：E:\logstash-7.16.2-windows-x86_64\logstash-7.16.2
然后再logstash的config目录下创建一个test.conf文件

1. input {
   
2.   stdin {
   
3.   }
   
4.   jdbc {
   
5.           type => 'testimport'
   
6.           jdbc_connection_string => 'jdbc:mysql://192.168.1.111:3306/weibo1?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull'
   
7.           jdbc_user => 'weibo1'
   
8.           jdbc_password => '123456'
   
9.           #开启记录最后一次运行的结果
   
10.           record_last_run => true
    
11.           #设置可以字段追踪
    
12.           use_column_value => true
    
13.           #指定数据库中用于追踪的字段
    
14.           tracking_column => 'id'
    
15.           #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
    
16.           tracking_column_type => 'numeric'
    
17.           #自定义的文件路径，用于存放追踪字段的最后导入的值
    
18.           last_run_metadata_path => 'e:/logstash-7.16.2-windows-x86_64/logstash-7.16.2/data/jdbc-position.txt'
    
19.           clean_run => 'false'
    
20.           #前提是在这个目录下放入mysql-connector-java-5.1.32.jar
    
21.           jdbc_driver_library => 'e:/logstash-7.16.2-windows-x86_64/logstash-7.16.2/mysql/mysql-connector-java-5.1.32.jar'
    
22.           jdbc_driver_class => 'com.mysql.jdbc.Driver'
    
23.           jdbc_paging_enabled => true
    
24.           jdbc_page_size => '500'
    
25.           statement => 'select * from baiduyuns where id > :sql_last_value'
    
26.           schedule => '* * * * *'
    
27.   }
    
28. }
    
29. filter {
    
30.         json {
    
31.                 source => 'message'
    
32.                 remove_field => ['message']
    
33.         }
    
34.         date {
    
35.                 match => ['timestamp','dd/MM/yyyy:HH:mm:ss Z']
    
36.         }
    
37. }
    
38. output {
    
39.         elasticsearch {
    
40.                #自己的es地址
    
41.                 hosts => ['http://121.199.65.242:9200']
    
42.                 index => 'testimport'
    
43.                 document_type => 'testimport'
    
44.         }
    
45.         stdout {
    
46.                 codec => json_lines
    
47.         }
    
48. }

复制代码

Windows开发环境下无法读取本地文件
报错代码：

1. [2019-11-25T19:55:47,213][ERROR][logstash.javapipeline    ][main]
   

复制代码

解决办法：下面这段是正确代码，注意是单引号！！！ 斜杆用“/”

1. input {  
   
2.     stdin{}
   
3.     file{
   
4.         path=>'C:/Users/Administrator/Desktop/test.txt'
   
5.     }
   
6. }

复制代码

把.conf  文件转成utf8格式


进入logstash的bin目录打开cmd执行命令：

1. logstash.bat -f /logstash-7.16.2-windows-x86_64/logstash-7.16.2/config/test.conf

复制代码

报错

1. “java.sql.SQLException: Value '0000-00-00' can not be represented as java.sql.Timestamp

复制代码

的错误， 这是因为hibernate认为这个不是一个有效的时间字串。

而有效的日期格式为“ 0001-01-01   00:00:00.0 ”

查看了mysql5的帮助文档对于datetime的解释如下

Datetimes with all-zero components (0000-00-00 ...) — These values can not be represented 关于所有Datetime类型由0组成的数据，这些值不能在java中被可靠的表示
reliably in Java.
Connector/J 3.0.x always converted them to NULL when being read from a ResultSet.
当这些值正在从ResultSet容器中读取时候，Connector/J 3.0.x 一直把他们转换为NULL值。

Connector/J 3.1 throws an exception by default when these values are encountered as this is the most correct behavior according to the JDBC and SQL standards.
依照JDBC和SQL的标准这些值碰到的最正确的处理方式就是在缺省情况下产生异常
This behavior can be modified using the zeroDateTimeBehavior configuration property. The allowable values are:
JDBC允许用下列的值对zeroDateTimeBehavior 属性来设置这些处理方式，

exception (the default), which throws an SQLException with an SQLState of S1009.
设置为exception 异常（缺省）用一个SQLState的s1009错误号来抛出一个异常
convertToNull, which returns NULL instead of the date.
设置为convertToNull，用NULL值来代替这个日期类型
round, which rounds the date to the nearest closest value which is 0001-01-01.
设置为round，则围绕这个日期最接近的值（0001-01-01）来代替

修改你的jdbc连接

1. jdbc:mysql://localhost/schoolmis?useUnicode=true&characterEncoding=utf8&<span style="box-sizing: border-box; outline: 0px; color: red;">zeroDateTimeBehavior=convertToNull</span>

复制代码