Logstash介绍

Logstash是elastic技术栈中的一个技术。它是一个数据采集引擎，可以从数据库采集数据到es中。我们可以通过设置自增id主键或者时间来控制数据的自动同步，这个id或者时间就是用于给logstash进行识别的

  • id：假设现在有1000条数据，Logstatsh识别后会进行一次同步，同步完会记录这个id为1000，以后数据库新增数据，那么id会一直累加，Logstatsh会有定时任务，发现有id大于1000了，则增量加入到es中
  • 时间：同理，一开始同步1000条数据，每条数据都有一个字段，为time，初次同步完毕后，记录这个time，下次同步的时候进行时间比对，如果有超过这个时间的，那么就可以做同步，这里可以同步新增数据，或者修改元数据，因为同一条数据的时间更改会被识别，而id则不会。

 

插件 logstash-input-jdbc

本插件用于同步，es6.x起自带，这个是集成在了 logstash中的。所以直接配置同步数据库的配置文件即可

 

安装Logstash

注：使用Logstatsh的版本号与elasticsearch版本号需要保持一致

注：记得安装JDK，java -version检查一下，如果没有安装，请安装一下。（JDK快速安装：https://www.sudo.ren/article/87?t=1586915815628）

1.下载Logstash-6.4.3.tar.gz到/application/tools，然后解压到/application

[root@elasticsearch tools]# ll
total 435900
-rw-r--r-- 1 root root  97872736 Apr 13 16:03 elasticsearch-6.4.3.tar.gz
-rw-r--r-- 1 root root 194545143 Dec 17 06:29 jdk-8u241-linux-x64.tar.gz
-rw-r--r-- 1 root root 153936585 Apr 15 09:27 logstash-6.4.3.tar.gz
[root@elasticsearch tools]# tar -xf logstash-6.4.3.tar.gz 
[root@elasticsearch tools]# ll
total 435900
-rw-r--r--  1 root root  97872736 Apr 13 16:03 elasticsearch-6.4.3.tar.gz
-rw-r--r--  1 root root 194545143 Dec 17 06:29 jdk-8u241-linux-x64.tar.gz
drwxr-xr-x 12 root root       255 Apr 15 09:29 logstash-6.4.3
-rw-r--r--  1 root root 153936585 Apr 15 09:27 logstash-6.4.3.tar.gz
[root@elasticsearch tools]# mv logstash-6.4.3 ../
[root@elasticsearch tools]# cd ..
[root@elasticsearch application]# ll
total 0
drwxr-xr-x  9 root  root  155 Apr 13 16:04 elasticsearch-6.4.3
lrwxrwxrwx  1 root  root   26 Apr 13 16:11 jdk -> /application/jdk1.8.0_241/
drwxr-xr-x  7 10143 10143 245 Dec 11 18:39 jdk1.8.0_241
drwxr-xr-x 12 root  root  255 Apr 15 09:29 logstash-6.4.3
drwxr-xr-x  2 root  root  103 Apr 15 09:30 tools


2.创建sync目录，并上传jdbc.jar（根据数据库版本），然后创建配置文件logstash-db-sync.conf

[root@elasticsearch logstash-6.4.3]# mkdir sync
[root@elasticsearch logstash-6.4.3]# cd sync/
[root@elasticsearch sync]# rz -E
rz waiting to receive.
-rw-r--r-- 1 root root 2293144 Jan 16 09:43 mysql-connector-java-8.0.16.jar
[root@elasticsearch sync]# 
[root@elasticsearch sync]# vim logstash-db-sync.conf

logstash-db-sync.conf内容如下：

input {
    jdbc {
        # 设置 MySql/MariaDB 数据库url以及数据库名称
        jdbc_connection_string => "jdbc:mysql://192.168.2.115:3306/foodie-shop?characterEncoding=utf8&zeroDateTimeBehavior=CONVERT_TO_NULL&serverTimezone=UTC&nullNamePatternMatchesAll=true&autoReconnect=true&useSSL=false"
        # 用户名和密码
        jdbc_user => "root"
        jdbc_password => "123456"
        # 数据库驱动所在位置，可以是绝对路径或者相对路径
        jdbc_driver_library => "/application/logstash-6.4.3/sync/mysql-connector-java-8.0.16.jar"
        # 驱动类名
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        # 开启分页
        jdbc_paging_enabled => "true"
        # 分页每页数量，可以自定义
        jdbc_page_size => "10000"
        # 执行的sql文件路径
        statement_filepath => "/application/logstash-6.4.3/sync/foodie-items.sql"
        # 设置定时任务间隔  含义：分、时、天、月、年，全部为*默认含义为每分钟跑一次任务
        schedule => "* * * * *"
        # 索引类型
        type => "_doc"
        # 是否开启记录上次追踪的结果，也就是上次更新的时间，这个会记录到 last_run_metadata_path 的文件
        use_column_value => true
        # 记录上一次追踪的结果值（不用手动创建，系统将自动生成）
        last_run_metadata_path => "/application/logstash-6.4.3/sync/track_time"
        # 如果 use_column_value 为true， 配置本参数，追踪的 column 名，可以是自增id或者时间
        tracking_column => "updated_time"
        # tracking_column 对应字段的类型
        tracking_column_type => "timestamp"
        # 是否清除 last_run_metadata_path 的记录，true则每次都从头开始查询所有的数据库记录
        clean_run => false
        # 数据库字段名称大写转小写
        lowercase_column_names => false
    }
}
output {
    elasticsearch {
        # es地址
        hosts => ["localhost:9200"]
        # 同步的索引名
        index => "foodie-items"
        # 设置_docID和数据相同
        document_id => "%{id}"
    }
    # 日志输出
    stdout {
        codec => json_lines
    }
}


3.创建sql脚本

SELECT
    i.id as itemId,
    i.item_name as itemName,
    i.sell_counts as sellCounts,
    ii.url as imgUrl,
    tempSpec.price_discount as price,
    i.updated_time as updated_time
FROM
    items i
LEFT JOIN
    items_img ii
on
    i.id = ii.item_id
LEFT JOIN
    (SELECT item_id,MIN(price_discount) as price_discount from items_spec GROUP BY item_id) tempSpec
on
    i.id = tempSpec.item_id
WHERE
    ii.is_main = 1
    and
    i.updated_time >= :sql_last_value

## :sql_last_value 表示Logstash的边界时间

修改数据时间

## 将数据推迟4个月
update items set updated_time = date_add(updated_time,INTERVAL 4 MONTH);



4.创建索引foodie-items



5.启动logstatsh

bin/logstash -f /application/logstash-6.4.3/sync/logstash-db-sync.conf


6.浏览数据

此时我们就成功将数据库中的数据同步到Elasticsearch上，接下来，正式开始表演。
 

## 通过查看/application/logstash-6.4.3/sync/track_time 可知其内容是一个时间字符串，凡是数据库中大于这时间都可以被Logstash同步到ES中

[root@elasticsearch sync]# cat track_time 
--- 2020-01-09 22:45:38.000000000 +08:00
[root@elasticsearch sync]# 

 

最后修改于 2020-04-15 09:51:57
如果觉得我的文章对你有用，请随意赞赏
扫一扫支付
上一篇