logstash接入Mysql数据库表

mysql 作为成熟稳定的数据持久化解决方案&＃xff0c;广泛地应用在各种领域&＃xff0c;但是在数据分析方面稍有不足&＃xff0c;而 elasticsearch 作为数据分析领域的佼佼者&＃xff0c;刚好可以弥补这项不足&＃xff0c;而我们要做的只需要将 mysql 中的数据同步到 elasticsearch 中即可&＃xff0c;而 logstash 刚好就可以支持&＃xff0c;所有你需要做的只是写一个配置文件而已

logstash 获取

wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.3.zip
unzip logstash-6.2.3.zip && cd logstash-6.2.3

安装 jdbc 和 elasticsearch 插件

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

获取 jdbc mysql 驱动

查找 插件名称&＃xff1a;

安装&＃xff1a;

yum install mysql-connector-java

编写配置文件

logstash-input-jdbc

使用 logstash-input-jdbc 插件读取 mysql 的数据&＃xff0c;这个插件的工作原理比较简单&＃xff0c;就是定时执行一个 sql&＃xff0c;然后将 sql 执行的结果写入到流中&＃xff0c;增量获取的方式没有通过 binlog 方式同步&＃xff0c;而是用一个递增字段作为条件去查询&＃xff0c;每次都记录当前查询的位置&＃xff0c;由于递增的特性&＃xff0c;只需要查询比当前大的记录即可获取这段时间内的全部增量&＃xff0c;一般的递增字段有两种&＃xff0c;AUTO_INCREMENT 的主键 id 和 ON UPDATE CURRENT_TIMESTAMP 的 update_time 字段&＃xff0c;id 字段只适用于那种只有插入没有更新的表&＃xff0c;update_time 更加通用一些&＃xff0c;建议在 mysql 表设计的时候都增加一个 update_time字段

• jdbc_driver_library: jdbc mysql 驱动的路径&＃xff0c;在上一步中已经下载
• jdbc_driver_class: 驱动类的名字&＃xff0c;mysql 填 com.mysql.jdbc.Driver 就好了
• jdbc_connection_string: mysql 地址
• jdbc_user: mysql 用户
• jdbc_password: mysql 密码
• schedule: 执行 sql 时机&＃xff0c;类似 crontab 的调度
• statement: 要执行的 sql&＃xff0c;以 “:” 开头是定义的变量&＃xff0c;可以通过 parameters 来设置变量&＃xff0c;这里的 sql_last_value 是内置的变量&＃xff0c;表示上一次 sql 执行中 update_time 的值&＃xff0c;这里 update_time 条件是 >&＃61; 因为时间有可能相等&＃xff0c;没有等号可能会漏掉一些增量
• use_column_value: 使用递增列的值
• tracking_column_type: 递增字段的类型&＃xff0c;numeric 表示数值类型, timestamp 表示时间戳类型
• tracking_column: 递增字段的名称&＃xff0c;这里使用 update_time 这一列&＃xff0c;这列的类型是 timestamp
• last_run_metadata_path: 同步点文件&＃xff0c;这个文件记录了上次的同步点&＃xff0c;重启时会读取这个文件&＃xff0c;这个文件可以手动修改
• hosts: es 集群地址
• user: es 用户名
• password: es 密码
• index: 导入到 es 中的 index 名&＃xff0c;这里我直接设置成了 mysql 表的名字
• document_id: 导入到 es 中的文档 id&＃xff0c;这个需要设置成主键&＃xff0c;否则同一条记录更新后在 es 中会出现两条记录&＃xff0c;%{id} 表示引用 mysql 表中 id 字段的值

运行

把上面的代码保存到一个配置文件里面  msyql_sqles.yml&＃xff0c;执行下面命令即可

cd /usr/local/elk/logstash && bin/logstash -f config/conf.d/mysql_sales.yml

如果成功了会在标准输出输出执行的 sql 语句

其他问题

多表同步

一个 logstash 实例可以借助 pipelines 机制同步多个表&＃xff0c;只需要写多个配置文件就可以了&＃xff0c;假设我们有两个表 table1 和 table2&＃xff0c;对应两个配置文件 sync_table1.cfg 和 sync_table2.cfg

在 config/pipelines.yml 中配置

- pipeline.id: table1
path.config: "config/conf.d/sync_table1.yml"
- pipeline.id: table2
path.config: "config/conf.d/sync_table2.yml"

直接 bin/logstash 启动即可

&＃64;timestamp 字段

默认情况下 &＃64;timestamp 字段是 logstash-input-jdbc 添加的字段&＃xff0c;默认是当前时间&＃xff0c;这个字段在数据分析的时候非常有用&＃xff0c;但是有时候我们希望使用数据中的某些字段来指定这个字段&＃xff0c;这个时候可以使用 filter.date, 这个插件是专门用来设置 &＃64;timestamp 这个字段的

比如我有我希望用字段 timeslice 来表示 &＃64;timestamp&＃xff0c;timeslice 是一个字符串&＃xff0c;格式为 %Y%m%d%H%M

filter {date {match &＃61;> [ "timeslice", "yyyyMMddHHmm" ]timezone &＃61;> "Asia/Shanghai"}
}

把这一段配置加到 sync_table.cfg 中&＃xff0c;现在 &＃64;timestamp 和 timeslice 一致了