logstash来实现全量和增量将mysql数据导入到es

以你之姓@ 2023-07-02 14:25 16阅读 0赞

## 1 简介 ##

本文说明使用logstash来实现全量和增量将mysql数据导入到es中。每一分钟执行一次。有时效性高的要求可以提高频率。

**2.logstash同步mysql数据到elasticsearch**

logstash-plugin install logstash-output-elasticsearch
    logstash-plugin install logstash-input-jdbc

将mysql-connector-java-8.0.11.jar copy到logstash/bin/mysql目录下

### 2.1 时间维度 ###

1.以一定的时间间隔定时更新mysql数据到es。

2.配置如下

全量构建索引

（1）jdbc.conf

input {
        jdbc {
          jdbc_default_timezone => "Asia/Shanghai"
          # mysql 数据库链接,school-edu为数据库名
          jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/school-edu?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull"
          # 用户名和密码
          jdbc_user => "xxxxx"
          jdbc_password => "xxxxx"
          # 驱动
          jdbc_driver_library => "E:\linux\ELK\6.8.2\logstash\bin\mysql\mysql-connector-java-8.0.18.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
        # 执行的sql 文件路径+名称
          statement_filepath => "E:\linux\ELK\6.8.2\logstash\config\mysql\jdbc.sql"
          # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
        }
    }
    
    output {
        elasticsearch {
          # ES的IP地址及端口
            hosts => ["localhost:9200"]
          # 索引名称
            index => "teacher"
            user => "elastic"
            password => "changeme"
            document_type => "_doc"
          # 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号
            document_id => "%{id}"
        }
        stdout {
         # JSON格式输出
            codec => json_lines
        }
    }

(2)jdbc.sql

select id, `name`, intro, career, avatar, sort, upd_time
    from edu_teacher

增量构建索引

(1)jdbc-update.conf

input {
        jdbc {
          jdbc_default_timezone => "Asia/Shanghai"
          # mysql 数据库链接,school-edu为数据库名
          jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/school-edu?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull"
          # 用户名和密码
          jdbc_user => "xxxxxx"
          jdbc_password => "xxxxx"
          # 驱动
          jdbc_driver_library => "E:\linux\ELK\6.8.2\logstash\bin\mysql\mysql-connector-java-8.0.18.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          last_run_metadata_path => "E:\linux\ELK\6.8.2\logstash\config\mysql\last_value_meta.txt"
        # 执行的sql 文件路径+名称
          statement_filepath => "E:\linux\ELK\6.8.2\logstash\config\mysql\jdbc-update.sql"
          # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
        }
    }
    
    output {
        elasticsearch {
          # ES的IP地址及端口
            hosts => ["localhost:9200"]
          # 索引名称
            index => "teacher"
            user => "elastic"
            password => "changeme"
            document_type => "_doc"
          # 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号
            document_id => "%{id}"
        }
        stdout {
         # JSON格式输出
            codec => json_lines
        }
    }

(2)jdbc-update.sql

select id, `name`, intro, career, avatar, sort, upd_time
    from edu_teacher
    where upd_time >= :sql_last_value

(3)last\_value\_meta.txt

`2020-01-29 00:00:00`

全量构建

logstash -f E:\linux\ELK\6.8.2\logstash\config\mysql\jdbc.conf

增量构建

logstash -f E:\linux\ELK\6.8.2\logstash\config\mysql\jdbc-update.conf

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQxNzIyNzE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQxNzIyNzE_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200130171202789.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQxNzIyNzE=,size_16,color_FFFFFF,t_70