分享

Flume中同时使用Kafka Source和Kafka Sink的Topic覆盖问题

问题导读:
1.在Kafka Sink中是什么样?
2.在Kafka Source中是什么样?
3.如何解决?





解决方案:


如果在一个Flume Agent中同时使用Kafka Source和Kafka Sink来处理events,便会遇到Kafka Topic覆盖问题,具体表现为,Kafka Source可以正常从指定的Topic中读取数据,但在Kafka Sink中配置的目标Topic不起作用,数据仍然会被写入到Source中指定的Topic中。
比如:在Agent中的Kafka Source配置Topic为:
agent_myAgent.sources.kafkaSource.topic = sourceTopic
在Kafka Sink配置Topic为:
agent_myAgent.sinks.kafkaSink.topic = sinkTopic
你会发现,最后数据又被写入到sourceTopic中,而sinkTopic没有任何数据写入。

在Kafka Sink中

配置项官网文档说明如下:
属性名topic,默认值为default-flume-topic。
The topic in Kafka to which the messages will be published. If this parameter is configured, messages will be published to this topic. If the event header contains a “topic” field, the event will be published to that topic overriding the topic configured here。
如果event header中包含了key为”topic”的值,那么将会覆盖该属性配置。

在源码org.apache.flume.sink.kafka.KafkaSink.process()中,

[mw_shl_code=java,true]if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
            eventTopic = topic;
}[/mw_shl_code]

其中topic是从属性agent_myAgent.sinks.kafkaSink.topic = sinkTopic 中获取的属性值(如果没有配置,则使用默认topic名称)
topic = context.getString(KafkaSinkConstants.TOPIC,KafkaSinkConstants.DEFAULT_TOPIC);

即:先使用event header中key为”topic”的值作为sink的topic,如果event header中没有,才取属性中配置的topic。


在Kafka Source中

源码:org.apache.flume.source.kafka.KafkaSource.process()



[mw_shl_code=java,true]// Add headers to event (topic, timestamp, and key)
    headers = new HashMap<String, String>();
    headers.put(KafkaSourceConstants.TIMESTAMP,
            String.valueOf(System.currentTimeMillis()));
    headers.put(KafkaSourceConstants.TOPIC, topic);[/mw_shl_code]

在event中,将Kafka Source中配置的topic加入到了header中。
因此,在Kafka Sink中,首先从event header中读取到了topic,Sink端的配置项不起作用。


解决办法

使用Flume拦截器,修改event header中key=topic的值为目标topic,拦截器使用Static interceptor,配置如下:

[mw_shl_code=java,true]## Source 拦截器
agent_myAgent.sources.kafkaSource.interceptors = i1
agent_myAgent.sources.kafkaSource.interceptors.i1.type = static
agent_myAgent.sources.kafkaSource.interceptors.i1.key = topic
agent_myAgent.sources.kafkaSource.interceptors.i1.preserveExisting = false
agent_myAgent.sources.kafkaSource.interceptors.i1.value = sinkTopic[/mw_shl_code]

其中,要特别注意preserveExisting属性值需要设置为false,该属性设置如果event header中已经存在key=topic,是否保留原来的值,默认为true,如果为true,那么还是会保留原来的topic,你设置的将不会生效。

另外,Kafka Sink中可以不用再设置topic属性了,反正也没用。
agent_myAgent.sinks.kafkaSink.topic = sinkTopic  //不需要了


没找到任何评论,期待你打破沉寂

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条