Logstash发送MySQL数据到Elasticsearch

2022-10-04

运维 / ELK

字数统计: 578 | 阅读时长≈ 2 分钟

使用 JDBC input plugin

Logstash 发送 MySQL 数据到 Elasticsearch 可以通过其内置的 Jdbc input plugin 。

需要两个配置步骤

准备 MySQL java 客户端包；
配置 Logstash 的 jdbc pipline 配置文件；

1. 准备 MySQL java 客户端包

将 jar 包放置 logstash/lib/mysql/ 目录下。例如：

1
2
3

[root@elk1-2 logstash-8.4.1]# ll lib/mysql/
total 2268
-rw-r--r-- 1 es es 2321813 Oct  3 20:17 mysql-connector-java-8.0.17.jar

2. 配置 Logstash 的 jdbc pipline 配置文件

将配置好的 jdbc pipline 配置文件，放到 config/logstash.yml 配置文件中 path.config 参数对应的目录下。

path.config 参数的作用： Where to fetch the pipeline configuration for the main pipeline

自定义的 pipline 配置文件，需要分别配置 input 和 output 部分。

样例配置如下：

input {
    jdbc {
        jdbc_connection_string => "jdbc:mysql://10.12.0.57:3306/cestc_sandboxie?characterEncoding=UTF-8&useSSL=false"
        jdbc_driver_library => "/opt/logstash-8.4.1/lib/mysql/mysql-connector-java-8.0.17.jar"
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        jdbc_user => "root"
        jdbc_password => "xxx"
        jdbc_validate_connection => "true"
        # statement => "SELECT * FROM cestc_sandboxie.SS_YHXW_LOG WHERE  CJSJ >= :sql_last_value AND CJSJ <= :sql_last_value"
        statement => "SELECT * FROM cestc_sandboxie.SS_YHXW_LOG limit 20"

        lowercase_column_names => false
        # 是否记录上次执行结果，true 表示会将上次执行结果的 tracking_column 字段的值保存到 last_run_metadata_path 指定的文件中
        # record_last_run => true
        record_last_run => false
        # 需要记录查询结果某字段的值时，此字段为 true，否则默认 tracking_column 为 timestamp 的值
        # use_column_value => true
        use_column_value => false
        # 需要记录的字段，用于增量同步，数据表字段
        tracking_column => "CJSJ"
        # 增量同步的字段类型：
        tracking_column_type => timestamp
        # record_last_run 上次数据存放位置
        last_run_metadata_path => "/opt/logstash-8.4.1/data/mysql/last_id"
        # 是否清除 last_run_metadata_path 的记录，需要增量同步时此字段必须为 false
        clean_run => false
        # 同步频率，默认每分钟同步一次
        #schedule => "* * * * *"
    }
}
output {
  elasticsearch {
    hosts => ["http://10.12.1.84:9200"]
    index => "10.12.0.57-sandbox-yhxw-%{+YYYY.MM.dd}"
    # 数据的唯一索引，这里填写查询的表的主键或一个值唯一的字段
    document_id => "%{ID}"
  }
}

3. 重启 Logstash

1	/opt/logstash-8.4.1/bin/logstash &

服务会读取 path.config 参数目录下所有的 conf 文件。

参考链接

版权声明： 本博客所有文章除特别声明外，著作权归作者所有。转载请注明出处！