2022-9-2-【运维】基于elasticsearch和logstash同步mysql数据到es

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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# mysql_to_es.conf

input{

    # 实有人口
    jdbc {
	  type => "mysql-to-es"
      # mysql相关jdbc配置
      jdbc_connection_string => "jdbc:mysql://192.168.7.11:19001/grid_collection?useUnicode=true&characterEncoding=utf8&useSSL=false&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true"
      jdbc_user => "root"
      jdbc_password => "cc123456"
 
      # jdbc连接mysql驱动的文件
      #jdbc_driver_library => "/usr/share/logstash/logstash-core/lib/jars/mysql-connector-java-5.1.49.jar"
      jdbc_driver_library => "/usr/share/logstash/mysql/mysql-connector-java-5.1.49.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
	  jdbc_paging_mode => "explicit" # 手动分页
      jdbc_fetch_size => "2000" # 每次查2000后即插入ES，避免OOM
      jdbc_page_size => "2000"
	  
 
      # mysql文件, 也可以直接写SQL语句在此处，如下：
      statement_filepath => "/usr/share/logstash/mysql/mysql-to-es.sql"
 
      # 这里类似crontab,可以定制定时操作，比如每10分钟执行一次同步(分 时 天 月 年)
      schedule => "*/10 * * * * *"
      # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）
      lowercase_column_names => false
      # 处理中文乱码问题
      codec => plain { charset => "UTF-8" }
      # 使用其它字段追踪，而不是用时间
      use_column_value => true
      # 追踪的字段  这个字段只有在上面的lowercase_column_names配置为false的时候才会区分大小写  因为不配置为false的话默认是true  查询结果字段默认会变成全小写；   我这里使用的是更新时间
      tracking_column => "updatedTime"
      # 这个就是追踪字段的类型，只有数值和时间两个类型(numeric和timestamp，默认numeric) 这个值会记录到last_run_metadata_path 配置的文件中 如果配置是numeric 那么默认值为0 如果配置为timestamp 那么默认值为1970年
      tracking_column_type => "timestamp"
      # 记录上一次运行记录
      record_last_run => true
      # 上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值  这个就是增量数据同步的关键
      last_run_metadata_path => "/usr/share/logstash/mysql/ga_syfwdjxx_to_es.index"
      # 是否清除 last_run_metadata_path 的记录,如果为true那么每次都相当于从头开始查询所有的数据库记录
      clean_run => false
      # 这个type可以用来做多个输入源和多个输出源区分  这里举例所以只有一个
    }

}
filter{
			#if[type] == "loginlog"{
			#    grok {
			#		match => {"message" => "|%{GREEDYDATA:loginMsg}|%{GREEDYDATA:timeFormat}|%{GREEDYDATA:userName}"}
			#	}
			#	if([message] =~  "^(?!.*?登录系统).*$") {
			#		### 丢弃
			#		drop{}
			#	   }
			#	}
			#if[type] == "datalog"{	
			#	grok {
			#		match => {"message" => "|%{DATA:userName}|%{GREEDYDATA:operationName}|%{DATA:timeFormat}|%{DATA:ip}|%{DATA:systemType}|%{GREEDYDATA:logType}|%{GREEDYDATA:method}|%{GREEDYDATA:input}"}
			#	}	
			#}
            #ruby {
			#	code => "event['time'] = event['@timestamp']"
			#}
			#mutate
			#{
			#	add_field => ["time", "%{@timestamp}"]
			#}				 
}
output{

	if[type] == "mysql-to-es"{
		elasticsearch{
			hosts=>["elasticsearch:9200"]
			user => "elastic"
			password => "Az@123456"
			index => "mysql-to-es" # 插入的index
			#将"_id"的值设为mysql的主键
			document_type => "_doc"
			document_id => "%{id}"
			manage_template => true
			template_overwrite => true
      # 同步时创建的使用的mapping模板
			template_name => "mysql-to-es"
			template => "/usr/share/logstash/template/mysql-to-es.json"
		}
	}
}