我尝试只为德国（de）过滤kafka json消息。要做到这一点，我必须写一个grok表达式。有人能帮我为这个json写一个grok模式吗？

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

抱歉，我对这些技术还不熟悉。下面是我的logstash.conf的样子：

input { 
  kafka {topics => [ "test" ] auto_offset_reset => "earliest" } 
} 

filter { 
  grok {
    match => { "message" => "?????????" }

  if [message] =~ "*COUNTRY*DE*" { 
    drop{}
  }
}      
}

output { file { path => "./test.txt"  } }

最后我只想向德国提交订单。希望能得到一些帮助，谢谢！