当前位置：首页 > 编程日记 > 正文

日志文件和mysql同步到kafka_logstash_output_kafka:Mysql同步Kafka深入详解

编程日记 2024-10-04 01:20:00

0、题记

实际业务场景中，会遇到基础数据存在Mysql中，实时写入数据量比较大的情景。迁移至kafka是一种比较好的业务选型方案。

而mysql写入kafka的选型方案有：

方案一：logstash_output_kafka 插件。

方案二：kafka_connector。

方案三：debezium 插件。

方案四：flume。

方案五：其他类似方案。

其中：debezium和flume是基于mysql binlog实现的。

如果需要同步历史全量数据+实时更新数据，建议使用logstash。

1、logstash同步原理

常用的logstash的插件是：logstash_input_jdbc实现关系型数据库到Elasticsearch等的同步。

实际上，核心logstash的同步原理的掌握，有助于大家理解类似的各种库之间的同步。

logstash核心原理：输入生成事件，过滤器修改它们，输出将它们发送到其他地方。

logstash核心三部分组成：input、filter、output。

input { }

filter { }

output { }

1.1 input输入

包含但远不限于：

jdbc：关系型数据库：mysql、oracle等。

file：从文件系统上的文件读取。

syslog：在已知端口514上侦听syslog消息。

redis：redis消息。beats：处理 Beats发送的事件。

kafka：kafka实时数据流。

1.2 filter过滤器

过滤器是Logstash管道中的中间处理设备。您可以将过滤器与条件组合，以便在事件满足特定条件时对其执行操作。

可以把它比作数据处理的ETL环节。

一些有用的过滤包括：

grok：解析并构造任意文本。Grok是目前Logstash中将非结构化日志数据解析为结构化和可查询内容的最佳方式。有了内置于Logstash的120种模式，您很可能会找到满足您需求的模式！

mutate：对事件字段执行常规转换。您可以重命名，删除，替换和修改事件中的字段。

drop：完全删除事件，例如调试事件。

clone：制作事件的副本，可能添加或删除字段。

geoip：添加有关IP地址的地理位置的信息。

1.3 output输出

输出是Logstash管道的最后阶段。一些常用的输出包括：

elasticsearch：将事件数据发送到Elasticsearch。

file：将事件数据写入磁盘上的文件。

kafka：将事件写入Kafka。

详细的filter demo参考：http://t.cn/EaAt4zP

2、同步Mysql到kafka配置参考

input {

jdbc {

jdbc_connection_string => "jdbc:mysql://192.168.1.12:3306/news_base"

jdbc_user => "root"

jdbc_password => "xxxxxxx"

jdbc_driver_library => "/home/logstash-6.4.0/lib/mysql-connector-java-5.1.47.jar"

jdbc_driver_class => "com.mysql.jdbc.Driver"

#schedule => "* * * * *"

statement => "SELECT * from news_info WHERE id > :sql_last_value order by id"

use_column_value => true

tracking_column => "id"

tracking_column_type => "numeric"

record_last_run => true

last_run_metadata_path => "/home/logstash-6.4.0/sync_data/news_last_run"

}

filter {

ruby{

code => "event.set('gather_time_unix',event.get('gather_time').to_i*1000)"

}

ruby{

code => "event.set('publish_time_unix',event.get('publish_time').to_i*1000)"

}

mutate {

remove_field => [ "@version" ]

remove_field => [ "@timestamp" ]

remove_field => [ "gather_time" ]

remove_field => [ "publish_time" ]

}

output {

kafka {

bootstrap_servers => "192.168.1.13:9092"

codec => json_lines

topic_id => "mytopic"

}

file {

codec => json_lines

path => "/tmp/output_a.log"

}

以上内容不复杂，不做细讲。

注意：

Mysql借助logstash同步后，日期类型格式：“2019-04-20 13:55:53”已经被识别为日期格式。

code =>

"event.set('gather_time_unix',event.get('gather_time').to_i*1000)"，

是将Mysql中的时间格式转化为时间戳格式。

3、坑总结

3.1 坑1字段大小写问题

from星友：使用logstash同步mysql数据的，因为在jdbc.conf里面没有添加 lowercase_column_names

=> "false" 这个属性，所以logstash默认把查询结果的列明改为了小写，同步进了es，所以就导致es里面看到的字段名称全是小写。

最后总结：es是支持大写字段名称的，问题出在logstash没用好，需要在同步配置中加上 lowercase_column_names => "false" 。记录下来希望可以帮到更多人。

3.2 同步到ES中的数据会不会重复？

想将关系数据库的数据同步至ES中，如果在集群的多台服务器上同时启动logstash。

解读：实际项目中就是没用随机id 使用指定id作为es的_id ，指定id可以是url的md5.这样相同数据就会走更新覆盖以前数据

3.3 相同配置logstash,升级6.3之后不能同步数据。

解读：高版本基于时间增量有优化。

tracking_column_type => "timestamp"应该是需要指定标识为时间类型，默认为数字类型numeric

3.4 ETL字段统一在哪处理?

解读：可以logstash同步mysql的时候sql查询阶段处理，如：select a_value as avalue***。

或者filter阶段处理,mutate rename处理。

mutate {

rename => ["shortHostname", "hostname" ]

}

或者kafka阶段借助kafka stream处理。

4、小结

相关配置和同步都不复杂，复杂点往往在于filter阶段的解析还有logstash性能问题。

需要结合实际业务场景做深入的研究和性能分析。

有问题，欢迎留言讨论。

TPYBoard自制微信远程智能温湿度计

2019独角兽企业重金招聘Python工程师标准>>> 智能时代一夜间什么都能远程了。创业者想着如何做智能产品，如何做远程控制。DIY爱好者也想着如何自制各种奇妙的工具。这里和大家一起学习制作一款廉价的智能温湿度计。说它廉价是因为共计花费不过40元&#…

编程日记2024/10/04 01:10:00

Matlab数据的可视化 -- 简易线性函数图

本篇微信图文主要介绍Matlab数据可视化方面的内容。

编程日记2024/10/04 01:00:00

浏览器tab关闭事件_Python--使用Pyqt5实现简易浏览器（最新版本测试过）

Python--使用Pyqt5实现简易浏览器(最新版本测试过)博客说明文章所涉及的资料来自互联网整理和个人总结，意在于个人学习和经验汇总，如有什么地方侵权，请联系本人删除，谢谢！准备环境首先我们需要的是我们的开发环境&…

编程日记2024/10/04 00:50:00

JS 中 URL 编码的问题.

URL 编码为什么要对 URL 编码 1. 避免解析错误我们的 queryString 的形式是使用 ?开始, keyvalue 传递参数, key-value pairs 之间使用 & 连接.比如: ?postid5038412&t1450591802326服务器会根据 & 解析 key-value pairs 根据解析 key,value 那么如果 key或…

编程日记2024/10/04 00:40:00

通用权限管理模块系列——需求分析——列举需求

以RBAC为理论基础以低耦合为设计原则我们计划设计一个独立的权限模块碰到一些难题，后面另外开文章来讨论本文主要是列举目前收集到的权限方面的需求信息，如果各位有能提供的用例，请不吝赐教。各种需求都是从网络或零碎项目里提取的&…

编程日记2024/10/04 00:30:00

Matlab数据的可视化 -- 散点图

本篇微信图文主要介绍Matlab数据可视化方面的内容。scatter

编程日记2024/10/04 00:20:00

koa连接mysql_CentOS 环境 Node + Koa2 连接 MySQL （ECS系列三）

Koa 搭建服务新建一个文件夹 koa_server，进入后cnpm i -S koa会生成 node_modules 依赖包，以及 package.json创建文件 server.js，并写入const Koa require(koa);const app new Koa();app.use( async ( ctx ) > {ctx.body hello koa2})a…

编程日记2024/10/04 00:10:00

Python CRC32 文件校验

binascii.crc32(s [,crc])返回CRC32校验。参数crc指定初始值用于循环。例如：Code>>> import binascii>>> crc binascii.crc32(spam)>>> binascii.crc32( and eggs, crc)739139840>>> binascii.crc32(spam and eggs)739139840C…

编程日记2024/10/04 00:00:00

国家智慧城市战略实施保温材料等建材万亿市场待挖掘

近年来，国家多部委联合发布《国家新型城镇化规划(2014—2020)》《关于促进智慧城市健康发展的指导意见》，提出到2020年建成一批特色鲜明的智慧城市。国家智慧城市建设与发展上升为国家战略。2014年中国智慧城市的市场规模达到800多亿元，其中不…

编程日记2024/10/03 23:50:00

Matlab数据的可视化 -- 极坐标图及其与直角坐标图的转换

本篇微信图文主要介绍Matlab数据可视化方面的内容。polar、pol2cart、cart2pol

编程日记2024/10/03 23:40:00

mysql 主从复制性能_zabbix监控mysql各项性能，主从复制

######################################################监控mysql(默认监控模板不能用，再agentd.conf开启自定义key，自己编写脚本)在zabbix_agentd.conf尾部添加(注意相关文件路径，以及脚本权限等问题)vim zabbix_agentd.confUserParameter…

编程日记2024/10/03 23:30:00

SQL Server基础知识之:设计和实现视图

设计和实现视图可谓是数据库物理设计中的一个非常重要的步骤。从一般意义上说，设计和实现视图应该遵循下面的一些建议和原则。以下内容摘在文档，我对某些重点进行了补充说明（红色部分） 只能在当前数据库中创建视图。但是&#x…

编程日记2024/10/03 23:20:00

Matlab数据的可视化 -- 条形图

本篇微信图文主要介绍Matlab数据可视化方面的内容。bar、barh

编程日记2024/10/03 23:10:00

全球SDN测试认证中心发布OpenDaylight测试报告

随着软件定义网络(Software Defined Network, SDN)商业部署速度地加快,关乎整个SDN 网络性能表现的控制平面核心组件——SDN 控制器也越来越成为网络用户关心的焦点。日前，天地互连-全球SDN测试认证中心（SDNCTC，www.sdnctc.com）正…

编程日记2024/10/03 23:00:00

mysql可以做日期处理吗_mysql可以做日期处理吗

最近一个项目，需要写很多的sql语句。涉及到很多与时间相关的处理，现在做一下简单的总结。1.时间格式：在这个项目中，或许是由不同的人建立的数据库表结构吧，对时间的格式步统一，有的表中存储的long类型的时间…

编程日记2024/10/03 22:50:00

大型网站采用什么系统架构保证性能稳定性

from http://www.bobd.cn/design/web/Theory/200904/31145.html 千万级的注册用户，千万级的帖子，nTB级的附件，还有巨大的日访问量，大型网站采用什么系统架构保证性能和稳定性？ 首先讨论一下大型网站需要注意和考虑的问…

编程日记2024/10/03 22:40:00

Matlab数据的可视化 -- 饼图

本篇微信图文主要介绍Matlab数据可视化方面的内容。pie

编程日记2024/10/03 22:30:00

Rushcrm:企业部署CRM系统做什么

现在很流行的一个词就是大数据，很多企业开始重视数据的开发利用。CRM在这样的环境背景下诞生了，但这并不意味着所有的企业都能很好的利用好CRM系统(客户关系管理系统)。CRM带来的效益是显而易见的，但是购置了CRM系统仅仅是第一步，…

编程日记2024/10/03 22:20:00

slot没有毁灭的问题_解析flink之perjob模式下yn参数不生效问题

概要：0. 问题背景1. Stream Job的切分2. 计算资源的调度 & 任务的执行3. 最后的总结0. 问题背景：开始用flink处理流式作业的时候，用yarn-cluster模式提交作业的时候，脚本如下：$FLINK_BIN run -m yarn-cluster -yqu…

编程日记2024/10/03 22:10:00

就用到两条.net自带的获取数据库信息的语句 OleDbConnection con new OleDbConnection(connection); // OleDB数据库连接实例 // 获取数据库中表的相关信息DataTable dt con.GetOleDbSchemaTable(OleDbSchemaGuid.Tables, new Object[] { null, null, null, "Table"…

编程日记2024/10/03 22:00:00