关于MySQL导入数据到elasticsearch的小工具logstash

logstash核心配置文件pipelines.yml

 #注:此处的 - 必须顶格写必须！！！

 - pipeline.id: invitation

 #下面路径配置的是你同步数据是的字段映射关系

   path.config: /opt/apps/logstash/config/invitation/invitation.conf

同步数据时的字段映射关系配置文件invitation.conf。注：路径一定跟你pipelines.yml配置文件中的一样

 input {

 jdbc {

 #驱动jar包的位置

 jdbc_driver_library => "/opt/apps/logstash/lib/mysql-connector-java-8.0.13.jar"

 #驱动类名

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

 #MySQL的链接

 jdbc_connection_string => "jdbc:mysql://192.168.0.234:3306/community?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true"

 #数据库用户名

 jdbc_user => "mostchh"

 #数据库密码

 jdbc_password => "1qaz2wsx3edc"

 #数据库重连尝试次数

 connection_retry_attempts => "3"

 #超时时间

 jdbc_validation_timeout => "3600"

 #开启分页查询（默认false不开启）

 jdbc_paging_enabled => "true"

 #单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）

 jdbc_page_size => "500"

 #时区

 #jdbc_default_timezone =>"Asia/Shanghai"

 #如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；

 statement_filepath => "/opt/apps/logstash/sql/invitation/invitation.sql"

 #需要记录查询结果某字段的值时，此字段为true

 use_column_value => true

 #是否设置字段为小写，默认是true

 lowercase_column_names => false

 #需要记录的递增字段，用于增量同步，下次只同步比该值大的数据

 tracking_column => "modifiedTime"

 #递增的字段类型

 tracking_column_type => "timestamp"

 #记录上一次运行记录

 record_last_run => true

 #上一次同步的递增字段存放文件路径

 last_run_metadata_path => "/opt/apps/logstash/station/invitation.txt"

 #是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false

 clean_run => false

 #自动同步数据的cron表达式，下面是一秒执行一次

 schedule => "*/1 * * * * *"

 #对应你pipelines配置文件的ID

 type => "invitation"

 }

 }

 #数据处理的过滤器

 filter {

 aggregate {

 task_id => "%{cardId}"

 code => "

 	map['cardId'] = event.get('cardId')

 	map['title'] = event.get('title')

                 map['content'] = event.get('content')

                 map['issueUserId'] = event.get('issueUserId')

                 map['issueUserName'] = event.get('issueUserName')

                 map['issueUserIcon'] = event.get('issueUserIcon')

 	map['issueTime'] = event.get('issueTime')

 	map['revealStatus'] = event.get('revealStatus')

                 map['commentNum'] = event.get('commentNum')

                 map['isMeLike'] = event.get('isMeLike')

                 map['giveLikeNum'] = event.get('giveLikeNum')

                 map['isDelete'] = event.get('isDelete')

 				map['issueStatus'] = event.get('issueStatus')

 				map['cardStatus'] = event.get('cardStatus')

 	map['giveLikeUsers'] ||=[]

 	#数据一对多的处理

 	if (event.get('userId') != nil)

 		if !(map['giveLikeUsers'].include? event.get('userId'))

 			map['giveLikeUsers'] << event.get('userId')

 		end

 	end

 	map['file_list'] ||=[]

 	map['fileList'] ||=[]

 	#数据一对多的处理

 	if (event.get('fileId') != nil)

 		if !(map['file_list'].include? event.get('fileId'))

 			map['file_list'] << event.get('fileId')

 			map['fileList'] << {

 				'fileId' => event.get('fileId'),

 				'fileName' => event.get('fileName'),

 				'fileUrl' => event.get('fileUrl')

 			}

 		end

 	end

 	event.cancel()

 "

 push_previous_map_as_event => true

 timeout => 5

 }

 mutate {

 }

  mutate {

  #过滤不需要的字段

remove_field => ["@timestamp","@version"]

 }

 }

 output {

 elasticsearch {

document_id => "%{cardId}"

document_type => "_doc"

index => "bbs_card_management"

hosts => ["http://192.168.0.178:9200"]

  }

 stdout{

codec => rubydebug

 }

 }

同步数据的SQL配置文件invitation.sql，具体的SQL就根据你的业务来定了，我这里用的视图所以SQL比较简单。注：路径一定跟你invitation.conf配置文件中的一样
```
 SELECT

 *

 FROM

 invitation

 WHERE

 modifiedTime >= :sql_last_value

 AND modifiedTime < NOW()
```
此处的:sql_last_value 取得就是你递增字段存放地址中的值
递增字段存放文件invitation.txt。注：路径一定跟你invitation.conf配置文件中的一样
```
 --- 2021-08-30 15:22:08.000000000 +00:00
```
配置的存放类型是时间类型，存储格式就是这样的。

以上就是logstash的所有配置了，只需要运行即可实现一秒同步一次数据，当然具体多久同步根据你具体的需求来定。

秒客网

关于MySQL导入数据到elasticsearch的小工具logstash

相关文章