解决Maxwell发送Kafka消息数据倾斜问题

Posted piperck

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了解决Maxwell发送Kafka消息数据倾斜问题相关的知识,希望对你有一定的参考价值。

最近用Maxwell解析mysql的Binlog,发送到Kafka进行处理,测试的时候发现一个问题,就是Kafka的Offset严重倾斜,三个partition,其中一个的offset已经快200万了,另外两个offset才不到两百。
Kafka数据倾斜的问题一般是由于生产者使用的Partition接口实现类对分区处理的问题,一般是对key做hash之后,对分区数取模。当出现数据倾斜时,小量任务耗时远高于其它任务,从而使得整体耗时过大,未能充分发挥分布式系统的并行计算优势(参考Apache Kafka 0.10 技术内幕:数据倾斜详解)。
而使用Maxwell解析MySQL的Binlog发送到Kafka的时候,生产者是Maxwell,那么数据倾斜的问题明细就是Maxwell引起的了。

 

Maxwell官网查文档(Producers:kafka-partitioning Maxwell’s Daemon)得知,在Maxwell没有配置的情况下,默认使用数据库名作为计算分区的key,并使用Java默认的hashcode算法进行计算:

  1. A binlog event‘s partition is determined by the selected hash function and hash string as follows
  2. | HASH_FUNCTION(HASH_STRING) % TOPIC.NUMBER_OF_PARTITIONS
  3. The HASH_FUNCTION is either java‘s hashCode or murmurhash3. The default HASH_FUNCTION
  4. is hashCode. Murmurhash3 may be set with the kafka_partition_hash option.
  5. …………
  6. The HASH_STRING may be (database, table, primary_key, column). The default HASH_STRING
  7. is the database. The partitioning field can be configured using the
  8. producer_partition_by option.

而在很多业务系统中,不同数据库的活跃度差异是很大的,主体业务的数据库操作频繁,产生的Binlog也就很多,而Maxwell默认使用数据库作为key进行hash,那么显而易见,Binglog的操作经常都被分到同一个分区里面了。

 

于是我们在Maxwell启动命令中加入对应参数即可,这里我选择了Rowkey作为分区key,同时选用murmurhash3
哈希算法,以获得更好的效率和分布:

nohup /opt/maxwell-1.11.0/bin/maxwell --user=maxwell --password=*** --host=***
--exclude_dbs=/^(mysql|maxwell|test)/ --producer=kafka --kafka.bootstrap.servers=***
--kafka_partition_hash=murmur3 --producer_partition_by=primary_key >> /root/maxwell.log &

用此命令重新启动Maxwell之后,观察Offset的变化,隔一段时间之后,各分区Offset的增量基本一致,问题解决!

 

Reference:

https://leibnizhu.gitlab.io/2018/01/03/%E8%A7%A3%E5%86%B3Maxwell%E5%8F%91%E9%80%81Kafka%E6%B6%88%E6%81%AF%E6%95%B0%E6%8D%AE%E5%80%BE%E6%96%9C%E9%97%AE%E9%A2%98/index.html 转发此文 解决Maxwell发送Kafka消息数据倾斜问题

http://ningg.top/apache-kafka-10-best-practice-tips-data-skew-details/  Apache Kafka 0.10 技术内幕:数据倾斜详解

 




以上是关于解决Maxwell发送Kafka消息数据倾斜问题的主要内容,如果未能解决你的问题,请参考以下文章

maxwell收集binlog,不同网络环境通过nginx发送数据到kafka集群

kafka kafka如何设置指定分区进行发送和消费

Maxwell编译

MaxWell(Mysql实时数据采集)

Kafka数据倾斜到某一个分区解决方案

MySQL Binlog 解析工具 Maxwell 详解