11-Elasticsearch-logstash数据同步[Mysql->Logstash->Es]

logstash数据同步

简介

集中, 转换和存储数据, logstach是免费且开放的服务器端数据处理管道, 能够从多个来源采集数据,转换数据,然后将数据发送到您最喜欢的"存储库"中

官网

代码语言：javascript复制

https://www.elastic.co/cn/logstash/

下载地址

代码语言：javascript复制

https://www.elastic.co/cn/downloads/logstash

功能

数据采集
以id或update_time作为同步边界
logstash-input-jdbc 插件
在es中, 预先创建索引

logstash安装

上传MySQL的驱动包和logstatch的安装包

代码语言：javascript复制

# 解压
tar -zxvf logstash-6.4.3.tar.gz
# 移动
mv logstash-6.4.3 /usr/local/
# 进入
cd /usr/local/logstash-6.4.3/
# 创建文佳佳
mkdir sync
# 进入sync
cd sync/
# 将驱动包拷贝到当前目录
cp /home/software/mysql-connector-java-5.1.41.jar ./
# 创建同步配置文件
vi logstash-db-sync.conf

配置文件

代码语言：javascript复制

input {
    jdbc {
      # 设置MySQL数据库URL以及数据库名称
      jdbc_connection_string => "jdbc:mysql://192.168.246.1:3306/foodie-dev?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"
      # 用户名 密码
      jdbc_user => "root"
      jdbc_password => "123456"
      # 驱动包
      jdbc_driver_library => "/usr/local/logstash-6.4.3/sync/mysql-connector-java-5.1.41.jar"
      # 驱动类
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      # 开启分页
      jdbc_paging_enabled => true
      # 每页数据量
      jdbc_page_size => "10000"
      # 执行的SQL文件路径
      statement_filepath => "/usr/local/logstash-6.4.3/sync/foodie-items.sql"
    # 设置定时任务间隔 含义: 分, 时, 天, 月, 年, 全部为*, 每分钟跑一次
      schedule => "* * * * *"
      # 索引类型
      type => "_doc"
      # 是否开启记录上次追踪的结果, 也就是上次的更新时间, 这个会记录到last_run_metadata_path的文件
      use_column_value => true
      # 记录上一次追踪的结果值
      last_run_metadata_path => "/usr/local/logstash-6.4.3/sync/track_time"
      # 如果 use_column_value 为 true, 配置本参数, 追踪的column名 可以是自增id或者时间
      tracking_column => "update_time"
      # tracking_column 对应的字段类型
      tracking_column_type => "timestamp"
      # 是否清除 last_run_metadata_path 的记录, true 则每次都从头开始查询所有的数据库记录
      clear_run => false
      # 数据库字段名, 大写转小写
      lowercase_column_names => false
  }
}
output {
    elasticsearch {
      # es 地址
      hosts => ["192.168.247.142:9200"]
      # 同步的索引名
      index => "foodie-items"
      # 设置docid和数据库id相同
      document_id => "%{id}"
  }
    # 日志输出
    stdout {
      codec => json_lines
  }
}
vi foodie-items.sql
select a.*,a.update_time from items a where a.isMaon = 1 and a.update_time >= :sql_last_value

启动

代码语言：javascript复制

# 进入文件夹
cd /usr/local/logstash-6.4.3/bin
# 启动
./logstash -f /usr/local/logstash-6.4.3/sync/logstash-db-sync.conf

记录时间

Logstash 多表同步到ES

代码语言：javascript复制

https://www.jianshu.com/p/2abcd6892c31

自定义Mapping配置中文分词

从ES中获取logstash模板

代码语言：javascript复制

http://192.168.247.142:9200/_template/logstash

获取到之后修改

增加分词器设置

删除外侧的logstatch包裹

其他不需要改动

进入sync目录创建文件, 将修改后的内容粘贴到文件中

vi logstash-ik.json

修改 logstash-db-sync.conf 配置文件

在es配置中增加自定义模板配置

0 人点赞