ELK+MySQL出现大量重复记录问题处理

一、使用Logstash使用jdbc从MySQL读取数据操作

1.1 安装jdbc插件

jdbc默认已安装，如果没安装使用logstash-plugin安装即可（logstash-plugin在logstash的bin目录下）：

logstash-plugin install logstash-input-jdbc

1.2 下载mysql jdbc驱动

mysql最新版是8.0，但使用时报错，这里仍以5.1版本演示

wget https://cdn.mysql.com//Downloads/Connector-J/mysql-connector-java-5.1.47.zip
# 不推荐放/tmp目录，但这里为了与用户无关所以直接解压到/tmp目录
unzip mysql-connector-java-5.1.47.zip -d /tmp

1.3 配置logstash文件

配置如下项，如无意外mysql中数据即可同步到elasticsearch

input{
  jdbc {
    # jdbc:mysql://数据库ip:端口/数据库名
    jdbc_connection_string => "jdbc:mysql://10.10.6.91:3306/expdb"
    # 数据库用户名      
    jdbc_user => "root"
    # 数据库密码
    jdbc_password => "root"
    # mysql java驱动文件位置
    jdbc_driver_library => "/tmp/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar"
    # 驱动类名，这个一般不需要改
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    # 与crontab类似的格式，* * * * *基本隔十来秒就同步一次
    schedule => "* * * * *"
    # 同步时要执行的sql语句，语句最终的结果集同步到elasticsearch; 我这里只是简单将表中所有记录查出
    # 如果语句很多，可使用statement_file来指定sql语句文件
    statement => "select * from exploit_records"
  }
}

filter {

}

output{
  elasticsearch {
    hosts => ["localhost:9200"]
  }
}

二、记录重复问题

2.1 单日记录重复问题

2.1.1 问题描述

elasticsearch以_id项作为主键，只要传过来的数据中_id项的值尚未存在elasticsearch会接收该数据。

而_id项的值默认其实是数据到来时elasticsearch自己生成的然后添加上去的，所以在这种情况下_id是不可能重复的，即所有数据都会被来者不拒地接收。

logstash jdbc插件对数据库的数据的操作是依照计划（schedule）执行sql语句（statement），所有查出来的数据就统统按output发过去。

在这种模式下，每执行一次计划（schedule）所有记录就会被发往elasticsearch一次，而_id是由elasticsearch生成的不会重复，所以不同次计划（schedule）的同一条记录会被elasticsearch反复收录。

2.1.2 问题处理办法

同一条记录被反复收录这种情况一般都不是我们想要的，而要消除这种情况其关键是要将_id变成不是由elasticsearch自动生成，更确切地讲是要将_id变成人为可控地生成。

这种配置也不难，elasticsearch允计通过使用document_id来配置_id。

假设我们表的主键是edb_id，为了避免重复，我们要以表主键值赋值给document_id即可。此时1.3中配置修改如下：

input{
  jdbc {
    # jdbc:mysql://数据库ip:端口/数据库名
    jdbc_connection_string => "jdbc:mysql://10.10.6.91:3306/expdb"
    # 数据库用户名      
    jdbc_user => "root"
    # 数据库密码
    jdbc_password => "root"
    # mysql java驱动文件位置
    jdbc_driver_library => "/tmp/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar"
    # 驱动类名，这个一般不需要改
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    # 与crontab类似的格式，* * * * *基本隔十来秒就同步一次
    schedule => "* * * * *"
    # 同步时要执行的sql语句，语句最终的结果集同步到elasticsearch; 我这里只是简单将表中所有记录查出
    # 如果语句很多，可使用statement_file来指定sql语句文件
    statement => "select * from exploit_records"
  }
}

filter {

}

output{
  elasticsearch {
    hosts => ["localhost:9200"]
    # 相比1.3只多加了这一句
    # 将表主键edb_id字段的值赋值给elasticsearch的主键
    document_id => "%{edb_id}"
  }
}

2.1.3 多路径输入问题

由于我们这里只查询了一张表，所以2.2中的配置没有什么问题。但只果此时要多输入一张表这张表没有edb_id字段，那么按document_id => "%{edb_id}"取到的edb_id会为空，这张表的数据会因_id项值都为空而被视为重复项不为elasticsearch所收录。

对于这种情况，jdbc预留了type项，可将type项赋值为表名进行区分。

但如果表中原来就有type字段，那么给type项赋值将会覆盖type字段的值，这是我们所不想看到的。此时就不适合使用type项了，只能转而使用其他条件加以区分比如if [edb_id] != ""等等。

如果不是读表而是读其他东西，比如读txt文件处理思路是一样的，只要找到一个if条件让txt文件的内容会转到其相应output输出即可。

input{
  jdbc {
    # jdbc:mysql://数据库ip:端口/数据库名
    jdbc_connection_string => "jdbc:mysql://10.10.6.91:3306/expdb"
    # 数据库用户名      
    jdbc_user => "root"
    # 数据库密码
    jdbc_password => "root"
    # mysql java驱动文件位置
    jdbc_driver_library => "/tmp/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar"
    # 驱动类名，这个一般不需要改
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    # 与crontab类似的格式，* * * * *基本隔十来秒就同步一次
    schedule => "* * * * *"
    # 同步时要执行的sql语句，语句最终的结果集同步到elasticsearch; 我这里只是简单将表中所有记录查出
    # 如果语句很多，可使用statement_file来指定sql语句文件
    statement => "select * from exploit_records"
    # 将type项赋值为表名exploit_records
    type => "exploit_records"
  }
  # 多读一张表cve_records
  jdbc {
    jdbc_connection_string => "jdbc:mysql://10.10.6.91:3306/expdb"
    jdbc_user => "root"
    jdbc_password => "root"
    jdbc_driver_library => "/tmp/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    schedule => "* * * * *"
    statement => "select * from cve_records"
    # 将type项赋值为表名cve_records
    type => "cve_records"
  }
}

filter {

}

output{
  if [type] == "exploit_records"{
    elasticsearch {
      hosts => ["localhost:9200"]
      document_id => "%{edb_id}"
    }
  }
  if [type] == "cve_records"{
    elasticsearch {
      hosts => ["localhost:9200"]
      document_id => "%{cve}"
    }
  }
}

2.2 多日记录重复问题

2.2.1 问题描述

经过2.1的处理后在单日中已不会出现重复记录，但由于elasticsearch中将不同日期收到的记录存到不同index中，index不同即便是同id也不会被认为是同一条记录所以记录会被收复，有时这种不同日期的重复我们也是不希望有的。

其实更本质而言，并不是“elasticsearch中将不同日期收到的记录存到不同index中”，elasticsearch只是接收记录将记录存放到记录index字段所标识的index中。而logstash的elasticsearch插件默认生成的index字段值是"logstash-%{+YYYY.MM.dd}"。

我们要避免不同日期记录重复，其做法就是重写index值将其设置成一个不随日期变化的固定值。

2.2.2 配置示例

input{
  jdbc {
    # jdbc:mysql://数据库ip:端口/数据库名
    jdbc_connection_string => "jdbc:mysql://10.10.6.91:3306/expdb"
    # 数据库用户名      
    jdbc_user => "root"
    # 数据库密码
    jdbc_password => "root"
    # mysql java驱动文件位置
    jdbc_driver_library => "/tmp/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar"
    # 驱动类名，这个一般不需要改
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    # 与crontab类似的格式，* * * * *基本隔十来秒就同步一次
    schedule => "* * * * *"
    # 同步时要执行的sql语句，语句最终的结果集同步到elasticsearch; 我这里只是简单将表中所有记录查出
    # 如果语句很多，可使用statement_file来指定sql语句文件
    statement => "select * from exploit_records"
  }
}

filter {

}

output{
  elasticsearch {
    hosts => ["localhost:9200"]
    # 将记录id设为表关键字，避免单日重复
    document_id => "%{edb_id}"
    # 将index由默认的logstash-%{+YYYY.MM.dd}设置成固定值，避免多日重复
    index => "exploit_records"
  }
}

三、if-else语句写法

在第二大节中我们多处用到if语句，这里再更具体说明一下其写法。

官方说明见：https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals

if-else语句格式如下，和各语言的写法是类似的：

if EXPRESSION {
  ...
} else if EXPRESSION {
  ...
} else {
  ...
}

而表达示（EXPRESSION）进一步可细化成操作项、操作符、和操作值三种元素。

比如在我们前面的写法中[type]就是操作项，==就是操作符，"exploit_records"就是操作值；操作项和操作值的位和其他语言一样可以互换（本质就是一个比较）。

if [type] == "exploit_records"{
  elasticsearch {
    hosts => ["localhost:9200"]
    document_id => "%{edb_id}"
  }
}

操作项----所有项的获取都和上边type一样使用中括号括起来的形式。

操作值----就是我们自己设定的操作项的比较标准。另外如果操作项是整型那操作值也应该是整型。

操作符----

一元操作符（unary）：!
布尔操作符（boolean）：and, or, nand, xor
相等操作符（equality）： ==, !=, <, >, <=, >=
正则操作符（regexp）： =~, !~ (checks a pattern on the right against a string value on the left，操作项在左边正则操作值在右边)
包含操作符（inclusion）： in, not in

参考：

https://segmentfault.com/a/1190000014387486