重庆企业网站推广价格,朔州网站设计公司,二级造价工程师怎么注册,网站开发留言板代码一、需求
Flume向kafka发送数据时#xff0c;同一个flume发送到kafka的数据总是固定在某一个partition中。而业务需求是发送的数据在所有的partition平均分布 二、实现
Flume的官方文档#xff1a;
Kafka Sink uses the topic and key properties from the FlumeEvent hea…一、需求
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.typeorg.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
agent1.sources.nginxlogSource.interceptors.UUIDi1.headerNamekey
agent1.sources.nginxlogSource.interceptors.UUIDi1.preserveExistingfalse 三、出现的问题
由于网络抖动国外nginx机器连接不上国内的kafka集群的部分机器nginx机器的flume通道堵塞内存占有率高导致Nginx机器的cpu飙升100%持续几十台机器崩溃发送告警。
故障原因
flume添加了UUID拦截器UUID拦截器给Event的header添加了一个key值flume在发送到kafka中根据key指定了固定分区。由于网络抖动该kafka分区连接不上分区的所有数据发送失败回滚到channel通道失败数据还是以key指定的分区进行重新发送发送数据一直失败回滚channel通道直到机器崩溃故障发生。 四、总结
不要使用UUID拦截器进行固定的分区发送数据量大或者网络抖动容易导致机器崩溃。应该重新编写kafkaSinkflume在发送数据的时候随机生成一个key发送到不同的分区。就算失败回滚到channel通道也会发送到新的分区。
示例
KeyedMessageString, byte[] data new KeyedMessageString, byte[](eventTopic, UUID.randomUUID().toString(), eventBody);
messageList.add(data);
