# Filebeat 5.0 output to Kafka multiple topics **URL:** \ **Category:** Beats\ **Tags:** filebeat\ **Created:** [December 3, 2016, 9:38pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934 "2016-12-03T21:38:59Z")\ **Posts on this page:** 6\ **Page:** 1
**Author:** ![gearoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gearoid/32/13572_2.png) [@gearoid](https://discuss.elastic.co/u/gearoid)\ **Post date:** [December 3, 2016, 9:38pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/1 "2016-12-03T21:38:59Z")
I installed Filebeat 5.0 on my app server and have 3 Filebeat prospectors, each of the prospector are pointing to different log paths and output to one kafka topic called `myapp_applog` and everything works fine. **My Filebeat output configuration to one topic - Working** ``` output.kafka: # initial brokers for reading cluster metadata hosts: ["broker.1.ip.address:9092", "broker.2.ip.address:9092", "broker.3.ip.address:9092"] # message topic selection + partitioning topic: 'myapp_applog' partition.round_robin: reachable_only: false required_acks: 1 compression: gzip max_message_bytes: 1000000 ``` What I want to do is send each of the log files to seperate topics based on a condition see documentation section on [topics](https://www.elastic.co/guide/en/beats/filebeat/master/kafka-output.html#_topics). I have tried to do it but no data is been sent to any of the topics. Does anyone know why my condition does not match or it is correct. I can seem to find an example on how to correctly use the "topics topic condition". Here is my kafka output to muliple topics configuration. **Not Working** ``` output.kafka: # initial brokers for reading cluster metadata hosts: ["broker.1.ip.address:9092", "broker.2.ip.address:9092", "broker.3.ip.address:9092"] # message topic selection + partitioning topics: - topic: 'myapp_applog' when: equals: document_type: applog_myappapi - topic: 'myapp_applog_stats' when: equals: document_type: applog_myappapi_stats - topic: 'myapp_elblog' when: equals: document_type: elblog_myappapi partition.round_robin: reachable_only: false required_acks: 1 compression: gzip max_message_bytes: 1000000 ``` Here is full filebeat.yml configuration file. ``` ################### Filebeat Configuration Example ######################### ############################# Filebeat ###################################### filebeat.prospectors: # App logs - prospector - input_type: log paths: - /myapp/logs/myapp.log exclude_lines: [".+? INFO[^*].+", ".+? DEBUG[^*].+"] exclude_files: [".gz$", ".tmp"] fields: api: myappapi environment: STG ignore_older: 24h document_type: applog_myappapi scan_frequency: 1s # Multine on Timestamp, YYYY-MM-DD # https://www.elastic.co/guide/en/beats/filebeat/master/multiline-examples.html multiline: pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}' negate: true match: after max_lines: 500 timeout: 5s # Server Stats - prospector - input_type: log paths: - /myapp/logs/serverstats.log # Exclude messages with log level exclude_lines: [".+? ERROR[^*].+", ".+? DEBUG[^*].+"] exclude_files: [".gz$", ".tmp"] fields: api: myappapi environment: STG ignore_older: 24h document_type: applog_myappapi_stats scan_frequency: 1s # ELB prospector - input_type: log paths: - /var/log/httpd/elasticbeanstalk-access_log document_type: elblog_myappapi fields: api: myappapi environment: STG exclude_lines: [".+? INFO[^*].+", ".+? DEBUG[^*].+"] exclude_files: [".gz$", ".tmp"] ignore_older: 24h # 0s, it is done as often as possible. Default: 10s scan_frequency: 1s registry_file: /var/lib/filebeat/registry ############################# Output ########################################## # Configure what outputs to use when sending the data collected by the beat. # Multiple outputs may be used. output.kafka: # initial brokers for reading cluster metadata hosts: ["broker.1.ip.address:9092", "broker.2.ip.address:9092", "broker.3.ip.address:9092"] # message topic selection + partitioning topics: - topic: 'myapp_applog' when: equals: document_type: applog_myappapi - topic: 'myapp_applog_stats' when: equals: document_type: applog_myappapi_stats - topic: 'myapp_elblog' when: equals: document_type: elblog_myappapi partition.round_robin: reachable_only: false required_acks: 1 compression: gzip max_message_bytes: 1000000 ``` ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [December 5, 2016, 3:32pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/2 "2016-12-05T15:32:29Z")
The `document_type` per prospector becomes the event field `type`. That's why the filter won't match. Instead of conditionals consider using the format string like: ```auto filebeat.prospectors: - ... document_type: myapp_applog - ... document_type: myapp_applog_stats - ... document_type: myapp_elblog output.kafka: topic: '%{[type]}' # use document_type to set topic ``` btw. the `topic` field in conditionals also supports format strings. ---
**Author:** ![gearoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gearoid/32/13572_2.png) [@gearoid](https://discuss.elastic.co/u/gearoid)\ **Post date:** [December 5, 2016, 4:30pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/3 "2016-12-05T16:30:31Z")
> [@steffens](#): > > Instead of conditionals consider using the format string like: so the conditionals on muliple topics doesn't work. > [@steffens](#): > > output.kafka: > topic: '%{[type]}' # use document\_type to set topic I don't need to use topics - topic, it will pick up the log base on the type and send to that topic? Am I understanding correct? ---
**Author:** ![steffens](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/steffens/32/79630_2.png) [@steffens](https://discuss.elastic.co/u/steffens)\ **Post date:** [December 6, 2016, 12:20pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/4 "2016-12-06T12:20:16Z")
> so the conditionals on muliple topics doesn't work. conditionals do work. But it looks a little overkill. you have to replace `document_type` with `type` in your conditionals. It just looks like overkill maintaining the conditionals if format strings will do the trick. > I don't need to use topics - topic, it will pick up the log base on the type and send to that topic? The `topic` setting is a format string. Format strings can access fields in events. In the sample I'm setting the topic for each event by reading the `type` field, which you have already configured in `document_type`. See [Format String docs](https://www.elastic.co/guide/en/beats/libbeat/current/config-file-format-type.html#_format_string_sprintf) ---
**Author:** ![gearoid](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gearoid/32/13572_2.png) [@gearoid](https://discuss.elastic.co/u/gearoid)\ **Post date:** [December 6, 2016, 1:03pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/5 "2016-12-06T13:03:17Z")
OK, thanks for that. I will post back when get it working later to show if it worked ok. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [January 3, 2017, 1:03pm UTC](https://discuss.elastic.co/t/filebeat-5-0-output-to-kafka-multiple-topics/67934/6 "2017-01-03T13:03:19Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.