微信搜索bigdata029 | 邀请体验:数阅–数据管理、OLAP分析与可视化平台 | 订阅本站 | 赞助作者:赞助作者

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

Flume lxw1234@qq.com 2806℃ 0评论

如果在一个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没有任何数据写入。

经过DEBUG和分析,原因如下:

在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()中,

if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
            eventTopic = topic;
 }

其中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()

// 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);

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

解决办法

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

## 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

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

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

 

如果觉得本博客对您有帮助,请 赞助作者

转载请注明:lxw的大数据田地 » Flume中同时使用Kafka Source和Kafka Sink的Topic覆盖问题

喜欢 (4)
分享 (0)
发表我的评论
取消评论
表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址