1. Toy模板网首页
  2. > Toy博客

mysql同步数据到es之logstash

10月前 作者：代号：猿a 分类：Toy博客 阅读(44) 违法举报

这篇具有很好参考价值的文章主要介绍了mysql同步数据到es之logstash。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

mysql同步数据到es

常用两种方式

1.使用 logstash
如果是历史数据同步我们可以用logstash,最快同步频率每分钟一次,如果对时效性要求高,慎用

2.使用 canal
实时同步,本文章未演示

使用logstash进行同步

logstash 特性:

  1. 无需开发，仅需安装配置logstash即可；
  2. 凡是SQL可以实现的logstash均可以实现（本就是通过sql查询数据）
  3. 支持每次全量同步或按照特定字段（如递增ID、修改时间）增量同步；
  4. 同步频率可控，最快同步频率每分钟一次（如果对实效性要求较高，慎用）；
  5. 不支持被物理删除的数据同步物理删除ES中的数据（可在表设计中增加逻辑删除字段IsDelete标识数据删除）。

实现原理

定时查询数据库中数据,更新到es中

logstash实现步骤

1.下载安装

注意版本要和自己的es版本一致
下载地址https://www.elastic.co/cn/downloads/past-releases#logstash

2.配置
  1. 在bin同级目录下创建"mysql"文件夹
    mysql同步数据到es之logstash
  2. 在刚创建的" mysql" 文件夹下创建文件jdbc.conf , last_time.txt 和 放入mysql驱动jar
    mysql同步数据到es之logstash
  3. 配置jdbc.conf文件
    单表同步
input {
	stdin {}
	jdbc {
		type => "jdbc"
		 # 数据库连接地址
		jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/TestDB?characterEncoding=UTF-8&autoReconnect=true""
		 # 数据库连接账号密码；
		jdbc_user => "username"
		jdbc_password => "pwd"
		 # MySQL依赖包路径；
		jdbc_driver_library => "mysql/mysql-connector-java-5.1.34.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 => "500"
		 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
		 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
		 # statement_filepath => "mysql/jdbc.sql"
		 # 注意数据库对的时间查出来和es中的时间格式不一致,会导致插入es失败,需要进行时间格式转换
		statement => "SELECT t.id as id,t.`name` as name,t.num as num,t.create_by as createBy,DATE_FORMAT(t.create_time,'%Y-%m-%d %H:%i:%s') as createTime,t.update_by as updateBy,DATE_FORMAT(t.update_time,'%Y-%m-%d %H:%i:%s') as updateTime ,DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') as lastTime FROM product as t WHERE DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') >= DATE_FORMAT(:sql_last_value,'%Y-%m-%d %H:%i:%s') order by t.last_time asc"
		 # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => false
		 # 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 => "ModifyTime"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		tracking_column_type => timestamp
		 # record_last_run上次数据存放位置；
		last_run_metadata_path => "mysql/last_id.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"
	}
}
 
filter {
	json {
		source => "message"
		remove_field => ["message"]
	}
	# convert 字段类型转换，将字段TotalMoney数据类型改为float；
	mutate {
		convert => {
			"TotalMoney" => "float"
		}
	}
}
output {
	elasticsearch {
		 # host => "192.168.1.1"
		 # port => "9200"
		 # 配置ES集群地址
		hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
		 # 索引名字，必须小写
		index => "consumption"
		 # 数据唯一索引（建议使用数据库KeyID）
		document_id => "%{KeyId}"
	}
	stdout {
		codec => json_lines
	}
}

多表同步

input {
	stdin {}
	jdbc {
		 # 多表同步时，表类型区分，建议命名为“库名_表名”，每个jdbc模块需对应一个type；
		type => "TestDB_DetailTab"
		
		 # 其他配置此处省略，参考单表配置
		 # ...
		 # ...
		 # record_last_run上次数据存放位置；
		last_run_metadata_path => "mysql\last_id.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"
	}
	jdbc {
		 # 多表同步时，表类型区分，建议命名为“库名_表名”，每个jdbc模块需对应一个type；
		type => "TestDB_Tab2"
		# 多表同步时，last_run_metadata_path配置的路径应不一致，避免有影响；
		 # 其他配置此处省略
		 # ...
		 # ...
	}
}
 
filter {
	json {
		source => "message"
		remove_field => ["message"]
	}
}
 
output {
	# output模块的type需和jdbc模块的type一致
	if [type] == "TestDB_DetailTab" {
		elasticsearch {
			 # host => "192.168.1.1"
			 # port => "9200"
			 # 配置ES集群地址
			hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
			 # 索引名字，必须小写
			index => "detailtab1"
			 # 数据唯一索引（建议使用数据库KeyID）
			document_id => "%{KeyId}"
		}
	}
	if [type] == "TestDB_Tab2" {
		elasticsearch {
			# host => "192.168.1.1"
			# port => "9200"
			# 配置ES集群地址
			hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
			# 索引名字，必须小写
			index => "detailtab2"
			# 数据唯一索引（建议使用数据库KeyID）
			document_id => "%{KeyId}"
		}
	}
	stdout {
		codec => json_lines
	}
}
  1. 启动运行
    在bin同级目录下执行命令
    mysql同步数据到es之logstash 
【windows】bin\logstash.bat -f mysql\jdbc.conf
【linux】nohup ./bin/logstash -f mysql/jdbc_jx_moretable.conf &
案例

提前创建好es索引,我用的是easy-es,直接启动自动创建了

  1. 数据库+初始数据
CREATE TABLE `product` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `name` varchar(255) DEFAULT NULL COMMENT '名称',
  `num` int(10) DEFAULT NULL COMMENT '数量',
  `last_time` datetime DEFAULT NULL COMMENT '最后修改时间',
  `create_time` datetime DEFAULT NULL COMMENT '创建时间',
  `update_time` datetime DEFAULT NULL COMMENT '修改时间',
  `create_by` varchar(255) DEFAULT NULL COMMENT '创建者',
  `update_by` varchar(255) DEFAULT NULL COMMENT '修改者',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;


INSERT INTO `es`.`product`(`id`, `name`, `num`, `last_time`, `create_time`, `update_time`, `create_by`, `update_by`) VALUES (1, '香甜水蜜桃', 50, '2022-08-12 14:42:17', '2022-08-12 10:47:56', NULL, 'qts', NULL);
INSERT INTO `es`.`product`(`id`, `name`, `num`, `last_time`, `create_time`, `update_time`, `create_by`, `update_by`) VALUES (2, '红红的火龙果', 90, '2022-08-12 15:17:36', '2022-08-12 14:12:41', NULL, 'qts', NULL);
  1. es索引
{
    "product": {
        "aliases": {
            "ee_default_alias": {}
        },
        "mappings": {
            "properties": {
                "@timestamp": {
                    "type": "date"
                },
                "@version": {
                    "type": "text",
                    "fields": {
                        "keyword": {
                            "type": "keyword",
                            "ignore_above": 256
                        }
                    }
                },
                "createBy": {
                    "type": "keyword"
                },
                "createTime": {
                    "type": "date",
                    "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
                },
                "id": {
                    "type": "long"
                },
                "lastTime": {
                    "type": "date",
                    "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
                },
                "name": {
                    "type": "text",
                    "analyzer": "ik_smart",
                    "search_analyzer": "ik_max_word"
                },
                "num": {
                    "type": "integer"
                },
                "type": {
                    "type": "text",
                    "fields": {
                        "keyword": {
                            "type": "keyword",
                            "ignore_above": 256
                        }
                    }
                }
            }
        },
        "settings": {
            "index": {
                "routing": {
                    "allocation": {
                        "include": {
                            "_tier_preference": "data_content"
                        }
                    }
                },
                "number_of_shards": "1",
                "provided_name": "product",
                "creation_date": "1660287566474",
                "number_of_replicas": "1",
                "uuid": "ey2A7AYKQB2OBvBUN-fN3Q",
                "version": {
                    "created": "7140299"
                }
            }
        }
    }
}
  1. sql
SELECT
	t.id AS id,
	t.`name` AS NAME,
	t.num AS num,
	t.create_by AS createBy,
	DATE_FORMAT( t.create_time, '%Y-%m-%d %H:%i:%s' ) AS createTime,
	t.update_by AS updateBy,
	DATE_FORMAT( t.update_time, '%Y-%m-%d %H:%i:%s' ) AS updateTime,
	DATE_FORMAT( t.last_time, '%Y-%m-%d %H:%i:%s' ) AS lastTime 
FROM
	product AS t 
WHERE
	DATE_FORMAT( t.last_time, '%Y-%m-%d %H:%i:%s' ) >= DATE_FORMAT(:sql_last_value, '%Y-%m-%d %H:%i:%s' ) 
ORDER BY
	t.last_time ASC
  1. jdbc.conf
input {
	stdin {}
	jdbc {
		type => "jdbc"
		 # 数据库连接地址
		jdbc_connection_string => "jdbc:mysql://localhost:3306/es?characterEncoding=UTF-8&autoReconnect=true"
		 # 数据库连接账号密码；
		jdbc_user => "root"
		jdbc_password => "root"
		 # MySQL依赖包路径；
		jdbc_driver_library => "D:\java_dev_tool\logstash\logstash-7.14.2\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 => "500"
		 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
		 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
		 # statement_filepath => "mysql/jdbc.sql"
		statement => "SELECT t.id as id,t.`name` as name,t.num as num,t.create_by as createBy,DATE_FORMAT(t.create_time,'%Y-%m-%d %H:%i:%s') as createTime,t.update_by as updateBy,DATE_FORMAT(t.update_time,'%Y-%m-%d %H:%i:%s') as updateTime ,DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') as lastTime FROM product as t WHERE DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') >= DATE_FORMAT(:sql_last_value,'%Y-%m-%d %H:%i:%s') order by t.last_time asc"
		 # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => false
		 # 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 => "lastTime"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		tracking_column_type => timestamp
		 # record_last_run上次数据存放位置；
		last_run_metadata_path => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\last_time.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"
	}
}
 
filter {
	json {
		source => "message"
		remove_field => ["message"]
	}
	# convert 字段类型转换，将字段TotalMoney数据类型改为float；
	mutate {
		convert => {
			"TotalMoney" => "float"
		}
	}
}
output {
	elasticsearch {
		 # host => "localhost"
		 # port => "9200"
		 # 配置ES集群地址
		hosts => ["localhost:9200"]
		 # 索引名字，必须小写
		index => "product"
		 # 数据唯一索引（建议使用数据库KeyID）
		document_id => "%{id}"
	}
	stdout {
		codec => json_lines
	}
}
  1. start.bat 放在bin同级目录下
bin\logstash.bat -f mysql\jdbc.conf

mysql同步数据到es之logstash

  1. 双击 start.bat启动
    mysql同步数据到es之logstash

  2. 测试

http://localhost:9200/product/_search
  1. 结果
{
    "took": 177,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 2,
            "relation": "eq"
        },
        "max_score": 1.0,
        "hits": [
            {
                "_index": "product",
                "_type": "_doc",
                "_id": "2",
                "_score": 1.0,
                "_source": {
                    "id": 2,
                    "lastTime": "2022-08-12 15:17:36",
                    "name": "红红的火龙果",
                    "updateBy": null,
                    "@version": "1",
                    "@timestamp": "2022-08-12T07:56:00.108Z",
                    "type": "jdbc",
                    "createBy": "qts",
                    "createTime": "2022-08-12 14:12:41",
                    "updateTime": null,
                    "num": 90
                }
            },
            {
                "_index": "product",
                "_type": "_doc",
                "_id": "1",
                "_score": 1.0,
                "_source": {
                    "id": 1,
                    "lastTime": "2022-08-12 15:54:58",
                    "name": "香甜水蜜桃",
                    "updateBy": null,
                    "@version": "1",
                    "@timestamp": "2022-08-12T07:57:00.212Z",
                    "type": "jdbc",
                    "createBy": "qts",
                    "createTime": "2022-08-12 10:47:56",
                    "updateTime": null,
                    "num": 50
                }
            }
        ]
    }
}

补充

1. filter中封装对象中的嵌套参数

  1. 索引
{
    "document": {
        "aliases": {
            "ee_default_alias": {}
        },
        "mappings": {
            "properties": {
                "commentList": {    <= 对此字段进行封装
                    "type": "nested",        
                    "properties": {
                        "commentTitle": {
                            "type": "text",
                            "analyzer": "ik_smart",
                            "search_analyzer": "ik_max_word"
                        }
                    }
                },
                "content": {
                    "type": "text",
                    "analyzer": "ik_smart",
                    "search_analyzer": "ik_max_word"
                },
                "id": {
                    "type": "long"
                },
                "title": {
                    "type": "keyword"
                },
                "type": {
                    "type": "text",
                    "fields": {
                        "keyword": {
                            "type": "keyword",
                            "ignore_above": 256
                        }
                    }
                },
                "update_time": {
                    "type": "date"
                }
            }
        },
        "settings": {
            "index": {
                "routing": {
                    "allocation": {
                        "include": {
                            "_tier_preference": "data_content"
                        }
                    }
                },
                "number_of_shards": "1",
                "provided_name": "document",
                "creation_date": "1660543629625",
                "number_of_replicas": "1",
                "uuid": "4j4JB89zSiaUFkl-G0UG2g",
                "version": {
                    "created": "7140299"
                }
            }
        }
    }
}
  1. 对应filter操作
# 指定input中type为document的数据进行过滤操作
# 注意:如果不使用if判断,则多条数据输入时,每条中有一下 id 参数的都会进行封装并
filter {
	if [type] == "document"
	{
		#这里的  target  标签会对应 es 中 products 文档 的 skus 字段
		jdbc_streaming {
			jdbc_driver_library => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\mysql-connector-java-5.1.35.jar"
			jdbc_driver_class => "com.mysql.jdbc.Driver"
			jdbc_connection_string => "jdbc:mysql://localhost:3306/es?characterEncoding=UTF-8&autoReconnect=true"
			jdbc_user => "root"
			jdbc_password => "root"
			# sensor_identifier参数名,id 对应input的sql中的返回参数
			parameters => { "sensor_identifier" => "id"}
	
			#这里不能使用statement_filepath的方式引入sql文件，会报错
			#statement_filepath => "/etc/logstash/pipeline/sql/filter_sku.sql"
			# 通过父表ID对子表中数据进行查询
			statement => "SELECT commentTitle FROM comment as t WHERE doc_id = :sensor_identifier"
	
			#这个commentList对应 es 索引中的 commentList 字段,如果没有,则会自定创建默认类型
			target => "commentList"
		}
	}
}

2. filter去掉查询语句中自动生成的字段

filter {
	# 去掉无用的字段@timestamp和@version
	mutate {
		remove_field => ["@timestamp","@version"]
	}
}

3. filter转换字段类型

filter {
	# convert 字段类型转换，将字段TotalMoney数据类型改为float；
	mutate {
		convert => {
			"TotalMoney" => "float"
		}
	}
}

案例conf文件

多表同步,嵌套类型封装

input {
	stdin {}
	jdbc {
		type => "product"
		 # 数据库连接地址
		jdbc_connection_string => "jdbc:mysql://localhost:3306/es?characterEncoding=UTF-8&autoReconnect=true"
		 # 数据库连接账号密码；
		jdbc_user => "root"
		jdbc_password => "root"
		 # MySQL依赖包路径；
		jdbc_driver_library => "D:\java_dev_tool\logstash\logstash-7.14.2\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 => "500"
		 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
		 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
		 # statement_filepath => "mysql/jdbc.sql"
		 # statement => "SELECT t.id as id,t.`name` as name,t.num as num,t.create_by as createBy,DATE_FORMAT(t.create_time,'%Y-%m-%d %H:%i:%s') as createTime,t.update_by as updateBy,DATE_FORMAT(t.update_time,'%Y-%m-%d %H:%i:%s') as updateTime ,DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') as lastTime FROM product as t WHERE DATE_FORMAT(t.last_time,'%Y-%m-%d %H:%i:%s') >= DATE_FORMAT(:sql_last_value,'%Y-%m-%d %H:%i:%s') order by t.last_time asc"
		statement_filepath => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\jdbc.sql"		
		# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => false
		 # 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 => "lastTime"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		tracking_column_type => timestamp
		 # record_last_run上次数据存放位置；
		last_run_metadata_path => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\last_time.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"
	}
	jdbc {
		type => "document"
		 # 数据库连接地址
		jdbc_connection_string => "jdbc:mysql://localhost:3306/es?characterEncoding=UTF-8&autoReconnect=true"
		 # 数据库连接账号密码；
		jdbc_user => "root"
		jdbc_password => "root"
		 # MySQL依赖包路径；
		jdbc_driver_library => "D:\java_dev_tool\logstash\logstash-7.14.2\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 => "500"
		 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
		 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
		 # statement_filepath => "mysql/jdbc.sql"
		statement => "SELECT t.id,t.title,t.content,DATE_FORMAT(t.update_time,'%Y-%m-%d %H:%i:%s') as update_time FROM document as t WHERE DATE_FORMAT(t.update_time,'%Y-%m-%d %H:%i:%s') >= DATE_FORMAT(:sql_last_value,'%Y-%m-%d %H:%i:%s')"
		# statement_filepath => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\jdbc.sql"		
		# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
		lowercase_column_names => false
		 # 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 => "update_time"
		 # Value can be any of: numeric,timestamp，Default value is "numeric"
		tracking_column_type => timestamp
		 # record_last_run上次数据存放位置； ※※※ 此处的时间文件创建一个新的 ※※※
		last_run_metadata_path => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\doc_last_time.txt"
		 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
		schedule => "* * * * *"
	}
}
 
filter {
	对document中的数据特殊处理
	if [type] == "document"
	{
		#这里的  target  标签会对应 es 中 products 文档 的 skus 字段
		jdbc_streaming {
			jdbc_driver_library => "D:\java_dev_tool\logstash\logstash-7.14.2\mysql\mysql-connector-java-5.1.35.jar"
			jdbc_driver_class => "com.mysql.jdbc.Driver"
			jdbc_connection_string => "jdbc:mysql://localhost:3306/es?characterEncoding=UTF-8&autoReconnect=true"
			jdbc_user => "root"
			jdbc_password => "root"
			parameters => { "sensor_identifier" => "id"}

			#这里不能使用statement_filepath的方式引入sql文件，会报错
			#statement_filepath => "/etc/logstash/pipeline/sql/filter_sku.sql"

			statement => "SELECT commentTitle FROM comment as t WHERE doc_id = :sensor_identifier"

			#这个skus对应 es 索引中的 skus字段
			target => "commentList"
		}
	}
	json {
		source => "message"
		remove_field => ["message"]
	}
	# convert 字段类型转换，将字段TotalMoney数据类型改为float；
	# 去掉无用的字段@timestamp和@version
	mutate {
		remove_field => ["@timestamp","@version"]
	}
}
output {
	if [type] == "product"
	{
		elasticsearch {
			 # host => "localhost"
			 # port => "9200"
			 # 配置ES集群地址
			hosts => ["localhost:9200"]
			 # 索引名字，必须小写
			index => "product"
			 # 数据唯一索引（建议使用数据库KeyID）
			document_id => "%{id}"
		}
	}
	if [type] == "document"
	{
		elasticsearch {
			 # host => "localhost"
			 # port => "9200"
			 # 配置ES集群地址
			hosts => ["localhost:9200"]
			 # 索引名字，必须小写
			index => "document"
			 # 数据唯一索引（建议使用数据库KeyID）
			document_id => "%{id}"
		}
	}
	stdout {
		codec => json_lines
	}
}

补充2

配置说明文档地址

input plugin

https://www.elastic.co/guide/en/logstash/current/input-plugins.html

filter plugin

https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

output plugin

https://www.elastic.co/guide/en/logstash/current/output-plugins.html

参考文章

https://zxiaofan.blog.csdn.net/article/details/86708490?spm=1001.2101.3001.6650.2&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-2-86708490-blog-125497958.pc_relevant_multi_platform_whitelistv3&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-2-86708490-blog-125497958.pc_relevant_multi_platform_whitelistv3&utm_relevant_index=4文章来源地址https://www.toymoban.com/news/detail-402804.html

到了这里，关于mysql同步数据到es之logstash的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • 通过logstash实现mysql与es的双向数据同步

    通过logstash实现mysql与es的双向数据同步

    参考题目 一种基于MySQL和Elasticsearch的数据同步方法及系统 基于MySQL和Elasticsearch的数据同步方法 一种基于MySQL和Elasticsearch的数据同步系统 基于MySQL和Elasticsearch的数据同步技术 目录 1【理论调研】 方案1：使用Logstash实现数据同步 方案2：使用Canal实现数据同步 方案3：使用Debe

    2024年02月15日
    浏览(33)
  • 使用Logstash同步mysql数据到Elasticsearch（亲自踩坑）_将mysql中的数据导入es搜索引擎利用logstash(1)

    使用Logstash同步mysql数据到Elasticsearch（亲自踩坑）_将mysql中的数据导入es搜索引擎利用logstash(1)

    先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7 深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！ 因此收集整理了一份《2024年最新大数据全套学习资料》，

    2024年04月28日
    浏览(52)
  • MySQL数据同步ES的常用思路和方法

    MySQL数据同步ES的常用思路和方法

    大家应该都在各种电商网站检索过商品，检索商品一般都是通过什么实现呢？搜索引擎Elasticsearch。 那么问题来了，商品上架，数据一般写入到MySQL的数据库中，那么用于检索的数据又是怎么同步到Elasticsearch的呢？ 这是能想到的最直接的方式，在写入MySQL，直接也同步往ES里写

    2023年04月14日
    浏览(31)
  • 使用logstash把mysql同步到es，Kibana可视化查看

    使用logstash把mysql同步到es，Kibana可视化查看

    Logstash下载地址：https://www.elastic.co/cn/downloads/logstash

    2024年02月02日
    浏览(42)
  • logstash同步数据从kafka到es集群

    logstash同步数据从kafka到es集群

    背景：需求是这样的，原始文件是txt文件（每天300个文件），最终想要的结果是每天将txt中的数据加载到es中，开始的想法是通过logstash加载数据到es中，但是对logstash不太熟悉，不知道怎么讲程序弄成读取一个txt文件到es中以后，就将这个txt原始文件备份并且删除掉，然后就想

    2024年02月15日
    浏览(42)

  • Logstash同步MySQL数据到ElasticSearch

    当MySQL数据到一定的数量级，而且索引不能实现时，查询就会变得非常缓慢，所以使用ElasticSearch来查询数据。本篇博客介绍使用Logstash同步MySQL数据到ElasticSearch，再进行查询。 测试环境 Windows系统 MySQL 5.7 Logstash 7.0.1 ElasticSearch 7.0.1 Kibana 7.0.1 ELK工具下载可访问:https://www.elastic

    2024年02月01日
    浏览(42)
  • 【elasticsearch专题】：Logstash从入门到同步MySQL数据

    【elasticsearch专题】：Logstash从入门到同步MySQL数据

      Elasticsearch是在数据处理生态系统中担任一个开源的分布式 搜索 和 分析 引擎的角色，专为 存储 、 检索 和分析大量的数据而打造。与此相伴的是Kibana，一个开源数据可视化平台，用于以优雅的方式展示Elasticsearch中的数据，并赋予用户创建仪表盘、图表和报告的能力。然

    2024年02月04日
    浏览(48)
  • 使用Logstash同步mysql数据到Elasticsearch（亲自踩坑）

    使用Logstash同步mysql数据到Elasticsearch（亲自踩坑）

    这篇文章主要介绍了如何使用Logstash同步mysql数据到Elasticsearch（亲自踩坑），如果帮助到了大家，希望用你毛茸茸的小手点个赞🤗；如有错误或未考虑周全的地方，希望在评论区留言🫡 Logstash官方文档提供了解决方案 一. 安装Logstash Logstash下载地址 下载版本一定要和Elastics

    2024年04月09日
    浏览(54)

  • Logstash 同步MySQL数据 至 开启HTTPS的ElasticSearch

    最近因为一些原因被迫学习了一周多的ElasticSearch，记录一下遇到的小问题。 此文件我理解为是Logstash的 * 可编译文件 *，我们通过编写此文件然后运行logstash去编译执行来让我们的数据按照自身期望的去传输。该文件主要包括input、filter和output三个部分，其中input和output是必要

    2024年02月01日
    浏览(38)
  • Lostash同步Mysql数据到ElasticSearch（二）logstash脚本配置和常见坑点

    Lostash同步Mysql数据到ElasticSearch（二）logstash脚本配置和常见坑点

    新建脚本文件夹 cd /usr/local/logstash mkdir sql cd sql vim 表名称.conf #如: znyw_data_gkb_logstash.conf 建立文件夹，保存资源文件更新Id mkdir -p /data/logstash/data/last_run_metadata 脚本JDBC插件参数说明： 第一步要做的事情，配置logstasht同步脚本文件，内容如下： 第二部要做的事情，根据_mapping.

    2024年02月07日
    浏览(46)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包