# How can I extract form the logged JSON message a field value that can be used as partition key to the JSON messages to a Kafka topic with filebeat?

**URL:** <https://discuss.elastic.co/t/how-can-i-extract-form-the-logged-json-message-a-field-value-that-can-be-used-as-partition-key-to-the-json-messages-to-a-kafka-topic-with-filebeat/220224>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [February 20, 2020, 4:20pm UTC](https://discuss.elastic.co/t/how-can-i-extract-form-the-logged-json-message-a-field-value-that-can-be-used-as-partition-key-to-the-json-messages-to-a-kafka-topic-with-filebeat/220224 "2020-02-20T16:20:09Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Avi\_Chalbani](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/avi_chalbani/32/63051_2.png) [@Avi\_Chalbani](https://discuss.elastic.co/u/Avi_Chalbani)\
**Post date:** [February 20, 2020, 4:20pm UTC](https://discuss.elastic.co/t/how-can-i-extract-form-the-logged-json-message-a-field-value-that-can-be-used-as-partition-key-to-the-json-messages-to-a-kafka-topic-with-filebeat/220224/1 "2020-02-20T16:20:09Z")

</div>

How can I extract form the logged JSON message a field value that can be used as partition key to the JSON messages to a Kafka topic with filebeat?

I have an event (JSON message) in file, that needs to be sent to Kafka over **filebeat**.  
A JSON message looks like this:  
{"time":1582213700.001,"interval":"2s","worker":11,"application":"1.1.1.1"}

I would like to send this message to Kafka. The partion key should be the application field in the JSON event message. How can I provide the custom application field within the JSON message as partion key for the Kafka record?

The filebeat.yml like this:  
…  
output.kafka:  
version: 0.10.1  
hosts: ["{KAFKA1}:9092", "{KAFKA2}:9092", "{KAFKA3}:9092"]  
topic: '%{[log\_topic]}'  
codec.format:  
string: '%{[message]}'  
key: '%{[message.application]:default}'  
partition.hash:  
hash:   
random: true # if false non-hashable events will be dropped  
required\_acks: 1  
compression: none

[https://www.elastic.co/guide/en/beats/libbeat/6.8/config-file-format-type.html#\_format\_string\_sprintf](https://www.elastic.co/guide/en/beats/libbeat/6.8/config-file-format-type.html#_format_string_sprintf)  
According to this reference we can refer to event field values by using the format string spec.

With this configuration the default message 'default' is always reported as key. How can I configure the filebeat.yml to extract the custom application field and use this information as Kafka partion key?

Furthermore I have tried to define a filed in the input section as follows:  
type: log  
enabled: true  
paths:  
- /var/logs/\*event.log  
fields:  
log\_topic: "event"  
application: '%{[application]} string'  
fields\_under\_root: true

and a corresponding kafka output as:  
output.kafka:  
version: 0.10.1  
hosts: ["{KAFKA1}:9092", "{KAFKA2}:9092", "{KAFKA3}:9092"]  
topic: '%{[log\_topic]}'  
codec.format:  
string: '%{[message]}'  
key: '%{[application]:default}'

```
  partition.hash:
    hash: []
    random: true # if false non-hashable events will be dropped
  required_acks: 1

```

But then the kafka partion key was always: %{[application]} string

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [March 19, 2020, 4:20pm UTC](https://discuss.elastic.co/t/how-can-i-extract-form-the-logged-json-message-a-field-value-that-can-be-used-as-partition-key-to-the-json-messages-to-a-kafka-topic-with-filebeat/220224/2 "2020-03-19T16:20:11Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.