感谢作者郭晓云投稿！
前言：公司（某银行旗下第三方支付平台）最近在做运维大数据项目，需要将各个监控系统的实时采集数据汇总到大数据平台进行智能告警和根因定位，Zabbix作为整个公司数据量最大的监控系统，超过12000的nvps，每周约产生400G左右的监控数据，如何将Zabbix的实时监控数据抽取出来并且不影响到Zabbix的性能？
之前做过一小部分的数据通过Zabbix API的方式获取，大量的数据肯定不行的。我们的目标是：在不影响Zabbix性能的前提下，将Zabbix的实时采样数据以标准格式输出。经过一系列的调研和测试，最终采用如下方案。（由于公司安全限制，部分代码无法分享，还请见谅。）
1. 技术架构：
在此技术架构中，开源软件Maxwell可以读取mysql的binlog文件存放于kafka，读取原理类似于mysql的存库方式，这样对数据库的性能消耗较低；采用python脚本将Zabbix的配置信息读取之后存放于Redis中，供后续Spark进行数据处理和组合；Spark算子会将kafka中的数据进行处理再和Redis的数据进行合并，最终获取到要求的标准数据；influxdb作为时序数据库大量存储标准化之后的数据；再将标准数据存放于另外一个kafka topic供运维大数据消费和计算。
2. Maxwell
Maxwell是一个能实时读取MySQL二进制日志binlog，并生成 JSON 格式的消息，作为生产者发送给 Kafka，Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序。相同功能的还有阿里巴巴开源的canal和Yelp开源的mysql_streamer，在使用过程中主要测试了canal和Maxwell，最终选定了Maxwell，对比如下：
- Maxwell没有canal那种server client模式，只有一个server把数据发送到消息队列或redis。如果需要多个实例，通过指定不同配置文件启动多个进程。
- Maxwell有一个亮点功能，就是canal只能抓取最新数据，对已存在的历史数据没有办法处理。而Maxwell有一个bootstrap功能，可以直接引导出完整的历史数据用于初始化，非常好用。
- Maxwell不能直接支持HA，但是它支持断点还原，即错误解决后重启继续上次点儿读取数据。
- Maxwell只支持Json格式，而Canal如果用Server client模式的话，可以自定义格式。
- Maxwell比Canal更加轻量级。
在部署过程中，通过修改Maxwell的配置，可以只同步history_uint、history_str、history_txt、history等四张表的数据。
原binlog里面的数据样式为：
代码语言：javascript复制insert into history_uint (itemid,clock,ns,value) values(40319,1614680424,22205258,1)通过Maxwell同步到kafka里的json数据为：
代码语言：javascript复制{"database":"zabbix","table":"history_uint","type":"insert","ts":1614680424,"xid":10595,"commit":true,"data":{"itemid":40319,"clock":1614680424,"ns":22205258,"value":1}}3. Redis
在通过Maxwell拿到的数据只有{"itemid":40319,"clock":1614680424,"ns":22205258,"value":1}，这个数据距离标准输出数据还缺少主机名、主机IP、监控项名称、Key等内容，这些内容无论通过itemid使用API查询还是直接查询数据库，所带来的QPS都非常高，会极大的影响数据库的性能，所以在这里使用Redis，通过如下SQL将监控项的全量数据查询出来，再进行数据处理，同步到Redis中，供给Spark进行查询。
代码语言：javascript复制SELECT
i.itemid,
i.NAME,
i.key_,
h.host,
it.ip
FROM
items i,
hosts h,
interface it
WHERE
i.hostid = h.hostid
AND h.hostid = it.hostid在Redis数据存储中，以Itemid为Key，主机名称、主机IP、监控项名称、监控项key为value，在存储的时候依旧存储为jason的字符串格式，方便Spark进行数据计算。
代码语言：javascript复制redis 127.0.0.1:6379> SET itemid_40319 {"name":"ICMP ping","Key_":"icmpping","HOST":"zabbix server","IP":"127.0.0.l"}
OK
redis 127.0.0.1:6379> GET itemid_40319
{"name":"ICMP ping","Key_":"icmpping","HOST":"zabbix server","IP":"127.0.0.l"}4. Spark
Spark承载的是原始数据的抽取和计算。将原始kafka中存放的数据提取为
{"itemid":40319,"clock":1614680424,"ns":22205258,"value":1}这样的有效数据，再通过Itemid去Redis拿到相应的配置信息进行整合，最终生成的数据格式为：
代码语言：javascript复制{"HOST":"zabbix server","IP":"127.0.0.l"，"name":"ICMP ping","Key_":"icmpping"，"itemid":40319,"clock":1614680424,"ns":22205258,"value":1}在实际操作过程中，Spark的算子程序采用了Python编写，此处的算子程序为整个架构最困难的地方，程序的执行效率，并发数量等等都会影响数据的计算效率，经过多次调试和大数据专家的专业分析，最终能够达到产线12000nvps的数据实时计算。整个规程还是极为不容易的。
5. 数据储存
Spark计算完成后的数据就是我们想要的标准数据了，一式两份，一份再反写到kafka的另一个topic，用以给到运维大数据做集中的数据分析；另外一份通过API写入到influxdb，做长时间的存储(生产zabbix采用的是18T的SSD盘数据库，数据存储时间不到12个月)，解决了之前的数据存储问题，也方便后续的数据读取。
6. 总结
在整个部署过程中，所采用的的Spark、Redis、Kafka公司均有公共基础组件和相应的技术支持，实施难度一般，Maxwell和influxdb的部署和研究较为繁琐，基本都是从头搞起，最终还是达到了预期的效果：在不影响Zabbix性能的前提下，将zabbix的实时采样数据以标准格式输出。后续就是准备开展各个监控系统的数据聚合模型、智能基线算法、单指标异常检测、根因定位的相关工作了。
