ES坑之logstash配置文件_logstash配置es

input {

stdin {

}

jdbc {

  # mysql 数据库链接

  jdbc_connection_string => "jdbc:mysql:localhost/database?characterEncoding=utf8"

  # 用户名和密码

  jdbc_user => "xxx"

  jdbc_password => "xxxx"

  # 驱动

  jdbc_driver_library => "D:/xx/xx/logstash-6.2.4/config/mysql-connector-java-8.0.18.jar"

  # 驱动类名

  jdbc_driver_class => "com.mysql.jdbc.Driver"

  jdbc_paging_enabled => "true"

  jdbc_page_size => "50000"

  # 执行的sql 文件路径+名称

  #statement_filepath => ""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

parameters => { “sql_last_value” => “UpdateTime” }

statement => “SELECT * FROM (SELECT * FROM table1 ) t WHERE t.updatetime > :sql_last_value”

# 设置监听间隔 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新

schedule => “* * * * *”

  # 索引类型

  #type => "article"

 # 防止自动将大小转为小写

  lowercase_column_names => false

  # 记录上一次运行记录

  record_last_run => true

  # 使用字段值

  use_column_value => true

 # 追踪字段名

  tracking_column => "updatetime"

  # 字段类型

  tracking_column_type => "timestamp"

 # 上一次运行元数据保存路径

  last_run_metadata_path => "./logstash_last_id"

 # 是否删除记录的数据

  clean_run => false

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

}

filter {

json {

    source => "message"

    remove_field => ["message"]

}
1
2
3
4
5
6
7

}

output {

elasticsearch {

hosts => "http://localhost:9200/"

index => "indexname"

     document_type => "articles"

document_id => "%{articleid}"

template_overwrite => true
1
2
3
4
5
6
7
8
9

}

这里输出调试，正式运行时可以注释掉

stdout {

  codec => json_lines
1

}

}

————————————————————————————

注意点

（1）jdbc_driver_library
mysql-connector-java-5.1.46.jar的存放目录，这个一定要配置正确，支持全路径和相对路径。如果配置不对，将会报“can ”错误。
（2）sql_last_value
标志目前logstash同步的位置信息（类似offset）。比如id、updatetime。logstash通过这个标志，可以判断目前同步到哪一条数据。
（3）statement、statement_filepath
statement：执行同步的sql语句，可以同步部分数据。
statement_filepath：存储执行同步的sql语句。不和statement同时使用。
（4）schedule
定时器，表示每隔多长时间同步一次数据。格式类似crontab。
（5）tracking_column、tracking_column_type
tracking_column：表示表中哪一列用于判断logstash同步的位置信息。与sql_last_value比较判断是否需要同步这条数据。
tracking_column_type：racking_column指定列的类型。支持两种类型：numeric（默认）、timestamp。注意：如果列是时间字段（比如updateTime），一定要指定这个类型为timestamp。我就踩了这个大坑。。。一直同步不成功！！！
（6）last_run_metadata_path
存储sql_last_value值的文件名称及位置。
（7）document_id
生成elasticsearch的文档值，尽量使用同步的数据中已有的唯一标识。比如同步订单数据，可以使用订单号。

————————————————————————————

在使用过程中遇到的一些问题：

1、数据同步，在同步数据时会有一个记录值，logstash会根据这个值来进行数据更新，我这里使用的是updateTime

根据内容修改时间进行更新数据，但是发现logstash记录的时间不是当地时间，具体时区是哪还没有发现（后续会补全这一块）

记录的时间总比数据库中的时间要快，所以导致数据总是更新不上。

在这一块目前做了几个尝试：

1.通过在sql语句中拼接和转换时间，试图将数据库中的时间转换为logstash记录中的数据时间

2.在logstash中增加filter配置

filter {
ruby {
code => “event.timestamp.time.localtime”
}
}
以上方法目前均没有实现，原因：

1.sql语句中添加时间函数后会报错，logstash执行语句时无法识别函数（没有找到原因）

2.按照网上找的方法，在logstash配置文件中添加filter后并没有起作用（没有找到原因）

这些问题需要进一步深入研究原因

2、logstash链接mysql数据库失败

报错一：[2020-07-25T00:21:08,797][ERROR][logstash.inputs.jdbc ] Unable to connect to database. Tried 1 times {:error_message=>“Java::JavaSql::SQLNonTransientConnectionException: Could not create connection to database server. Attempted reconnect 3 times. Giving up.”}

问题分析：

首先：因为部署环境为内网，防止是因为服务器之间通信问题，对mysql所在服务器做了Telnet测试，端口可以访问通，本机安装Navicat链接mysql成功，网络原因排除

随后：检查了多次链接字符串，因为mysql密码中含有“@”，“$”等特殊字符，修改了mysql密码做尝试，mysql方面的原因排除

随后：更换了jdbc版本，原使用版本为mysql-connector-java-8.0.19.jar,更换为mysql-connector-java-5.x.jar

最后请教大神，大神查阅资料后给出的建议是出错原因可能是时区问题，需要在mysql链接字符串中添加时区

将原本字符串

jdbc:mysql://172.20.21.10:3306/Project?characterEncoding=utf8&autoReconnect=true

更改为：

jdbc:mysql://localhost:3306/Project?useUnicode=true&characterEncoding=UTF-8&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC

然后更改了驱动名 jdbc_driver_class => “com.mysql.jdbc.Driver”

将其改为：com.mysql.cj.jdbc.Driver

修改后遂成功

在其他服务器没有出现以上链接出错的问题，具体原因暂未查明，在此记录