qidao123.com ToB IT社区-企服评测·应用市场»论坛 › 数据库 › 分布式数据库 › Logstash同步MySQL到ES

Logstash同步MySQL到ES

[复制链接]
发表于 2025-10-19 08:08:35 | 显示全部楼层 |阅读模式

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

×
Logstash同步MySQL到ES

条件

   软件版本
  

  • Logstash: 7.10.2
  • ES: 7.10.2
  • MySQL: 5.7
  Logstash和ES安装在Ｗindows中, MySQL安装在假造机中
  设置

user表建表语句

  1. CREATE TABLE `user` (
  2.   `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键',
  3.   `name` varchar(255) DEFAULT NULL,
  4.   `create_date` datetime DEFAULT NULL,
  5.   `update_date` datetime DEFAULT NULL,
  6.   PRIMARY KEY (`id`)
  7. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
复制代码
位置信息

创建userDTS_last_value.txt, 不必要内容, 到时间logstash会写入
同步user数据的userDTS.sql

  1. SELECT
  2.         id as id,
  3.         name as name,
  4.         DATE_FORMAT(create_date,'%Y-%m-%d %H:%i:%S') as createDate,
  5.         DATE_FORMAT(update_date,'%Y-%m-%d %H:%i:%S') as updateDate
  6. FROM
  7.         user
  8. WHERE
  9.         # 增量数据的查询条件，一条数据怎么才算增量数据就是由这个条件确定的，而其中的：:sql_last_value就是第2步中配置项中的last_run_metadata_path里配置的文件路径对应的文件中的值，也就是每次同步数据后最新一条的updateDate值，那么下次同步数据的时候，sql根据这个条件就不会查到非增量数据，同时第2步中配置项中的tracking_column配置的值就是updateDate，实际使用的时候可以根据场景更换这个字段
  10.         # 解决Logstash的sql_last_value 和MySQL时区不一致问题
  11.         update_date >= CONVERT_TZ(:sql_last_value, '+08:00', '+00:00')
  12.     AND CONVERT_TZ(:sql_last_value, '+08:00', '+00:00') < now()
复制代码
logstash-mysql-es.conf

   留意事项
  

  • MySQL驱动器jar包必要指定, 要和MySQL的版本兼容
  • logstash的同步不是实时同步的, 利用同步可以利用Canel, Flink CDC监听Binlog的方式, 可以参考博主的SpringBoot集成Flink-CDC
  • 存量更新和存量更新设置根据自己的需求开启
  • 相干位置改成自己, 好比mysql, sql存放位置
  • 启动logstash的时间必要指定logstash-mysql-es.conf设置启动

    • 我的设置存放logstash安装目次的config中, 进入bin目次, 实行cmd, 输入启下令为logstash -f …\config\logstash-mysql-es.conf

  1. # 输入源
  2. input {
  3.    jdbc {
  4.                 # 数据库连接参数
  5.                 jdbc_connection_string => "jdbc:mysql://192.168.132.10:3306/whitebrocade?characterEncoding=UTF-8&useUnicode=true&useSSL=false&serverTimezone=Asia/Shanghai"
  6.                 # mysql用户名
  7.                 jdbc_user => "root"
  8.                 # mysql密码
  9.                 jdbc_password => "12345678"
  10.                
  11.                 # mysql驱动器jar包, 我这里用的是5.7的MySQL
  12.                 jdbc_driver_library => "E:/software/Maven/repository/mysql/mysql-connector-java/5.1.49/mysql-connector-java-5.1.49.jar"
  13.                 # 驱动类名
  14.                 jdbc_driver_class => "com.mysql.jdbc.Driver"
  15.                 # 开启分页
  16.                 jdbc_paging_enabled => "true"
  17.                 # 最大页码
  18.                 jdbc_page_size => "1000"
  19.                
  20.                 # 这个type可以用来做多个输入源和多个输出源区分  这里举例所以只有一个
  21.                 type => "user_DTS"
  22.                 # 用于同步的查询sql
  23.                 statement_filepath => "E:/software/Logstash/test_data/userDTS.sql"
  24.                 # 直接书写sql语句
  25.                 # statement => "select * from user"
  26.                 # 加上jdbc时区, 要不然logstash的时间会不准确
  27.                 jdbc_default_timezone => "Asia/Shanghai"
  28.                 # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）
  29.                 lowercase_column_names => "false"
  30.                  # 处理中文乱码问题
  31.                 codec => plain { charset => "UTF-8" }
  32.                
  33.                 # 日志日志级别
  34.                 sql_log_level => warn
  35.                
  36.                 # 如果需要增量更新的话，则需要在input/jdbc下添加如下配置
  37.                 # 如果要使用其它字段追踪, 而不是用时间开启这个配置
  38.                 use_column_value => true
  39.                 # 这个就是追踪字段的类型，只有数值和时间两个类型(numeric和timestamp，默认numeric) 这个值会记录到last_run_metadata_path 配置的文件中 如果配置是numeric 那么默认值为0 如果配置为timestamp 那么默认值为1970年
  40.                 tracking_column_type => "timestamp"
  41.                 # 追踪的字段  这个字段只有在上面的lowercase_column_names配置为false的时候才会区分大小写  因为不配置为false的话默认是true  查询结果字段默认会变成全小写
  42.                 tracking_column => "updateDate"
  43.                 # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
  44.                 record_last_run => true
  45.                 # 上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值  这个就是增量数据同步的关键
  46.                 last_run_metadata_path => "E:/software/Logstash/test_data/userDTS_last_value.txt"
  47.                 # 是否清除 last_run_metadata_path 的记录,如果为true那么每次都相当于从头开始查询所有的数据库记录
  48.                 clean_run => false
  49.                 # 设置监听间隔  各字段含义秒（由左至右）分、时、天、月、年
  50.                 # 这里配置为每15秒扫描一次MySQL
  51.                 schedule => "*/15 * * * * *"
  53.     }
  54. }
  56. # 过滤
  57. filter {
  58.         mutate {
  59.                 # 去掉没用字段, 注意了type不要去除掉, 这里需要根据type判断是否同步到ES
  60.                 remove_field  => ["@version", "@timestamp"]
  61.         }
  62. }
  64. # 输出源
  65. output {
  66.         # 判断类型是否位user_DTS(即是否为user表相关的, 如果是就同步到ES中)
  67.     if[type] == "user_DTS"{
  68.                 elasticsearch {
  69.                         # elasticsearch url
  70.                         hosts => ["http://localhost:9200"]
  71.                         # 用户名&密码
  72.                         # user => "whitebrocade"
  73.                         # password => "whitebrocade"
  74.                         # 下面两个参数可以开启更新模式
  75.                         action => "update"
  76.                         doc_as_upsert => true
  77.                         # 索引名
  78.                         index => "user"
  79.                         # 文档id 设置成数据库的id
  80.                         document_id => "%{id}"
  81.        }
  82.     }
  83.     stdout {
  84.         # 以json格式输出到控制台, 用于调试使用, 正式运行时可以注释掉
  85.         codec => json_lines
  86.     }
  87. }
复制代码
ES创建user索引

Logstash同步的时间, 并不会创建user的索引, 以是这里我们得自己手动创建一下
   保举一个ES的客户端工具,es-clint, 下边利用这工具创建
  1. PUT /user
  2. {
  3.   "settings": {
  4.     "number_of_shards": 3,
  5.     "number_of_replicas": 2
  6.   },
  7.   "mappings": {
  8.     "properties": {
  9.       "id": {
  10.         "type": "long"
  11.       },
  12.       "name": {
  13.         "type": "text"
  14.       },
  15.       "createDate": {
  16.         "type": "date",
  17.         "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
  18.       },
  19.       "updateDate": {
  20.         "type": "date",
  21.         "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
  22.       }
  23.     }
  24.   }
  25. }
复制代码
创建效果


测试

如今的MySQL和ES都没有数据
新增

往MySQL中新建一条数据

  1. INSERT INTO user (name, create_date, update_date)
  2. VALUES ('xm', NOW(), NOW());
复制代码

logstash日记


检察ES的user索引


修改

修改id为1的数据

  1. UPDATE
  2.   user
  3. SET
  4.   name = 'whiteBrocade',
  5.   update_date = NOW()
  6. WHERE id = 1;
复制代码

logstash日记


检察ES的user索引


参考资料

Running Logstash on Docker | Logstash Reference8.17
Docker 安装 LogStash的详细过程
Docker摆设Logstash同步Mysql数据到ES方式
利用docker安装logstash的详细方法
回复

使用道具 举报

返回列表

写过一篇
+ 我要发帖

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表