# Filter Kafka JSON messages with Logstash grok

**URL:** <https://discuss.elastic.co/t/filter-kafka-json-messages-with-logstash-grok/107251>\
**Category:** Logstash\
**Created:** [November 10, 2017, 9:39pm UTC](https://discuss.elastic.co/t/filter-kafka-json-messages-with-logstash-grok/107251 "2017-11-10T21:39:57Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![Konstantin\_Blag](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/konstantin_blag/32/24090_2.png) [@Konstantin\_Blag](https://discuss.elastic.co/u/Konstantin_Blag)\
**Post date:** [November 10, 2017, 9:39pm UTC](https://discuss.elastic.co/t/filter-kafka-json-messages-with-logstash-grok/107251/1 "2017-11-10T21:39:57Z")

</div>

I try to filter kafka json messages only for one Country - Germany (DE). To do that I have to write a grok expression. Can anyone help me in writing a grok pattern for this json?

```
{"table":"ORDERS","type":"I","payload":{"ID":"28112","COUNTRY":"DE","AMT":15.36}}
{"table":"ORDERS","type":"I","payload":{"ID":"28114","COUNTRY":"US","AMT":25.75}}

```

sorry - I'm new to these technologies. Here is what my logstash.conf looks like:

```
input { kafka {topics => ["test"] auto_offset_reset => "earliest" } } 
filter { grok {
match => { "message" => "?????????" }
if [message] =~ "*COUNTRY*DE*" { drop{} }
       } }
output { file { path => "./test.txt" } }

```

First, I thought I could get it with

> filter { ... json { ...

but Kafka, sometimes releases several messages at one timestamp event as {"table"...}{"table"...}{"table"...} and it breaks the pipeline.

In the end I just wanna file with the Germany orders. Hope to get some help, thanks!

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [November 13, 2017, 6:43am UTC](https://discuss.elastic.co/t/filter-kafka-json-messages-with-logstash-grok/107251/2 "2017-11-13T06:43:53Z")

</div>

Don't use a grok filter to parse JSON. Use the json filter.

---

<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:** [December 11, 2017, 6:44am UTC](https://discuss.elastic.co/t/filter-kafka-json-messages-with-logstash-grok/107251/3 "2017-12-11T06:44:01Z")

</div>

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