Flume均匀发送数据到kafka的partition配置UUID Interceptor生成key的坑

àì夳堔傛蜴生んèń 2023-06-30 14:56 2阅读 0赞

### 一、需求 ###

Flume向kafka发送数据时，同一个flume发送到kafka的数据总是固定在某一个partition中。而业务需求是发送的数据在所有的partition平均分布

### 二、实现 ###

Flume的官方文档：

Kafka Sink uses the topic and key properties from the FlumeEvent headers to send events to Kafka. If topic exists in the headers, the event will be sent to that specific topic, overriding the topic configured for the Sink. If key exists in the headers, the key will used by Kafka to partition the data between the topic partitions. Events with same key will be sent to the same partition. If the key is null, events will be sent to random partitions.

kafka-sink是从header里的key参数的value值来hash到kafka的某个分区中。如果key为null，那么就会随机发布至分区中。事实上key为null被指定到kafka的某个固定分区。

要partition平均分布数据，就向header中写上随机的key，然后数据才会真正的向kafka分区进行随机发布。

官方文档有一个UUID Interceptor，会为每个event的head添加一个随机唯一的key，向flume添加拦截器达到随机分区发送。

在flume添加的配置文件如下：

agent1.sources.nginxlogSource.interceptors = UUIDi1
    agent1.sources.nginxlogSource.interceptors.UUIDi1.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
    agent1.sources.nginxlogSource.interceptors.UUIDi1.headerName=key
    agent1.sources.nginxlogSource.interceptors.UUIDi1.preserveExisting=false

### 三、出现的问题 ###

由于网络抖动，国外nginx机器连接不上国内的kafka集群的部分机器，nginx机器的flume通道堵塞，内存占有率高，导致Nginx机器的cpu飙升100%，持续几十台机器崩溃发送告警。

故障原因：

flume添加了UUID拦截器，UUID拦截器给Event的header添加了一个key值，flume在发送到kafka中根据key指定了固定分区。由于网络抖动，该kafka分区连接不上，分区的所有数据发送失败回滚到channel通道，失败数据还是以key指定的分区进行重新发送，发送数据一直失败回滚channel通道，直到机器崩溃故障发生。

### 四、总结 ###

不要使用UUID拦截器进行固定的分区发送，数据量大或者网络抖动容易导致机器崩溃。应该重新编写kafkaSink，flume在发送数据的时候随机生成一个key，发送到不同的分区。就算失败回滚到channel通道也会发送到新的分区。

示例：

KeyedMessage<String, byte[]> data = new KeyedMessage<String, byte[]>
              (eventTopic, UUID.randomUUID().toString(), eventBody);
    messageList.add(data);