提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

• 前言
• 一、logstash相关准备工作
• • 1.1 下载logstash
  • 1.2 下载JDBC相关驱动jar包
  • 1.3 创建logstash配置文件
• 二、使用步骤
• • 1.用指定配置文件启动logstash收集数据
• 总结

---

前言

工作涉及到了logstash把mysql表数据导入到es,但在ES的date 类型字段格式上和mysql时间格式字段不匹配而卡了很久，终于解决了，所以特地记录一下。

---

一、logstash相关准备工作

1.1 下载logstash

要下载和es对应版本的logstash: https://elasticsearch.cn/download/#seg-3
到Elasticsearch中文社区去下载，非常方便，也十分感谢该社区。

1.2 下载JDBC相关驱动jar包

mysql-connector-java.jar包的下载教程: https://www.jb51.net/article/190860.htm

1.进入mysql官网到下载页面

2. 拉倒最下方点击进入

3.点击到选择界面

4.选择/J

1.3 创建logstash配置文件

1.在bin同级目录下创建"mysql"文件夹

2.在刚创建的" mysql" 文件夹下创建文件jdbc.conf , last_time.txt 和 放入mysql驱动jar

配置jdbc.conf文件
此为单表同步

input {
	stdin {}
	jdbc {
		type => "jdbc"
		 # 数据库连接地址
		jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/ceshizhiwang?characterEncoding=UTF-8&autoReconnect=true"
		 # 数据库连接账号密码；
		jdbc_user => "root"
		jdbc_password => "root"
		 # MySQL依赖包路径；
		jdbc_driver_library => "F:\logstash-7.10.1\mysql\mysql-connector-java-5.1.35.jar"
		 # the name of the driver class for mysql
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		 # 数据库重连尝试次数
		connection_retry_attempts => "3"
		 # 判断数据库连接是否可用，默认false不开启
		jdbc_validate_connection => "true"
		 # 数据库连接可用校验超时时间，默认3600S
		jdbc_validation_timeout => "3600"
		 # 开启分页查询（默认false不开启）；
		jdbc_paging_enabled => "true"
		 # 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
		jdbc_page_size => "10000"
                codec => plain { charset => "UTF-8" }
		 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
		 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
		 # statement_filepath => "mysql/jdbc.sql"
		 # 注意数据库对的时间查出来和es中的时间格式不一致,会导致插入es失败,需要进行时间格式转换
		statement => "SELECT * FROM test order by create_date ASC"
		 # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => true
		# 设置字段字符格式，防止编码报错;
        columns_charset => {
                       "create_date" => "UTF-8"
         }
		 # Value can be any of: fatal,error,warn,info,debug，默认info；
		sql_log_level => warn
		 # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
		record_last_run => true
		 # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
		use_column_value => true
		 # 需要记录的字段，用于增量同步，需是数据库字段
		tracking_column => "create_date"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		tracking_column_type => timestamp
		 # record_last_run上次数据存放位置；
		last_run_metadata_path => "F:\logstash-7.10.1\mysql\last_time.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"

	}
}
 
#过滤器处理数据非常重要
filter {

#判断create_date内的值是否为空，不为空往下走
if ["create_date"]{
#ruby 工具 处理 create_date 数据
#第一步 拿到create_date的值
#第二步 补上缺少的8个小时时间
#第三步 转换成ES对应date类型字段的yyyy-MM-dd'T'HH:mm:ssZ 格式
#第四步 把处理好的时间替换原来的create_date的数据
   ruby {
        code => "
        timestamp = event.get('create_date')
        localtime = timestamp.time+28800
        localtimeStr = localtime.strftime('%Y-%m-%dT%H:%M:%S+0800')
        event.set('create_date', localtimeStr) #保存最新时间
        "
      }
    }
# 删除掉无用的 两个字段 ES索引不需要保存     
  mutate {
        remove_field => "@timestamp"
        remove_field => "@version"
   }
}

output {
	elasticsearch {
		 # host => "192.168.0.01"
		 # port => "9200"
		 # 配置ES集群地址
		hosts => ["192.168.0.01:9200"]
		 # 索引名字，必须小写
		index => "test_ceshizhiwang"
		 # 数据唯一索引（建议使用数据库KeyID）
		document_id => "%{id}"
	}
}

二、使用步骤

1.用指定配置文件启动logstash收集数据

在bin同级目录下执行命令

【windows】bin\logstash.bat -f mysql\jdbc.conf
【linux】nohup ./bin/logstash -f mysql/jdbc_jx_moretable.conf &

---

总结

这个logstash收集es数据属实搞了好久，最关键的就两个错误

1 错误: \xE5" from ASCII-8BIT to UTF-8", :error_class=>"LogStash::Json::GeneratorE

字面意思是只支持UTF-8的格式 字段并不是UTF-8 所以 在配置文件中必须用columns_charset 来指定下sql里面的字段 这个非常坑

2 错误: [create_date] with format [yyyy-MM-dd’T’HH:mm:ssZ||yyyy-MM-dd’T’HH:mm:ss]", “caused_by”=>{

字面意思是mysql表内这个create_date字段的时间格式和ES表内对应字段的时间格式------[yyyy-MM-dd’T’HH:mm:ssZ||yyyy-MM-dd’T’HH:mm:ss] 对不上，所以插入失败

这个我弄了很久 最后了解到用sql语句的 DATE_FORMAT()函数是达到不了这个格式效果的，所以必须使用过滤器处理数据再插入es表中。我首先使用的是 date 这个后面才了解，是处理@timestamp 的。所以我用了ruby 这个确实牛，完美的处理了数据转换了时间格式插入成功了，但其中走了很多弯路，因为网上面其他文章对这块的讲述有点模糊，好在最后成功了。

特此记录工作上学到的经验！希望生活越来越好！

薇拉小姐！！！！！！