Filter Kafka JSON messages with Logstash grok

Elastic Stack Logstash
1

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!

2

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

3

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