title: Publish Kafka outputs in Grafana
body: <p>Using Filebeat, I publish raw data in Kafka. I'm using NiFi to consume this data (using ConsumeKafka_2_6). The output is a list queue with 1 row (I need to do this because the information didn't maintain the order if I send one row for each line) that looks like this:</p>
  <pre><code>{
    &quot;@timestamp&quot;: &quot;2022-06-21T18:18:29.651Z&quot;,
    &quot;@metadata&quot;: {
      &quot;beat&quot;: &quot;filebeat&quot;,
      &quot;type&quot;: &quot;_doc&quot;,
      &quot;version&quot;: &quot;8.0.0&quot;
    },
    &quot;log&quot;: {
      &quot;offset&quot;: 3780214,
      &quot;file&quot;: {
        &quot;path&quot;: &quot;C:\\Program Files\\Filebeat\\inputs-test\\2022-05-23T13.07.27.0 - Copy (2).log&quot;
      }
    },
    &quot;message&quot;: &quot;@ 2022-05-23 12:25:39.879 [    30] [lap=126.18  uS]       Out of range&quot;,
    &quot;input&quot;: {
      &quot;type&quot;: &quot;filestream&quot;
    },
    &quot;agent&quot;: {
      &quot;version&quot;: &quot;8.0.0&quot;,
      &quot;ephemeral_id&quot;: &quot;0c4c1a87-11f1-4f75-8512-fa489cf455e0&quot;,
      &quot;id&quot;: &quot;552ab611-0ffb-44f4-83bf-77334c35950d&quot;,
      &quot;name&quot;: &quot;VM001&quot;,
      &quot;type&quot;: &quot;filebeat&quot;
    },
    &quot;ecs&quot;: {
      &quot;version&quot;: &quot;8.0.0&quot;
    },
    &quot;host&quot;: {
      &quot;name&quot;: &quot;VM001&quot;
    }
  }{
    &quot;@timestamp&quot;: &quot;2022-06-21T18:18:29.651Z&quot;,
    &quot;@metadata&quot;: {
      &quot;beat&quot;: &quot;filebeat&quot;,
      &quot;type&quot;: &quot;_doc&quot;,
      &quot;version&quot;: &quot;8.0.0&quot;
    },
    &quot;message&quot;: &quot;@ 2022-05-23 12:25:39.879 [    30] [lap=106.46  uS]       Zone finished unexpectdly. \n+Exception message: Zone is out of range\n+################################################\n+#   BEGIN OF STACK TRACE                       #\n+################################################\n+   Error \n+################################################\n+#   END OF STACK TRACE                         #\n+################################################&quot;,
    &quot;input&quot;: {
      &quot;type&quot;: &quot;filestream&quot;
    },
    &quot;host&quot;: {
      &quot;name&quot;: &quot;VM001&quot;
    },
    &quot;agent&quot;: {
      &quot;name&quot;: &quot;VM001&quot;,
      &quot;type&quot;: &quot;filebeat&quot;,
      &quot;version&quot;: &quot;8.0.0&quot;,
      &quot;ephemeral_id&quot;: &quot;0c4c1a87-11f1-4f75-8512-fa489cf455e0&quot;,
      &quot;id&quot;: &quot;552ab611-0ffb-44f4-83bf-77334c35950d&quot;
    },
    &quot;ecs&quot;: {
      &quot;version&quot;: &quot;8.0.0&quot;
    },
    &quot;log&quot;: {
      &quot;offset&quot;: 3780940,
      &quot;file&quot;: {
        &quot;path&quot;: &quot;C:\\Program Files\\Filebeat\\inputs-test\\2022-05-23T13.07.27.0 - Copy (2).log&quot;
      },
      &quot;flags&quot;: [
        &quot;multiline&quot;
      ]
    }
  }{
  </code></pre>
  <p>Now, I need to extract all the messages (in order) and apply a filter: only maintain the messages that say <em>out of range</em>.</p>
  <p>I tried to use <strong>EvaluateJsonPath</strong> like <a href="https://i.stack.imgur.com/quejP.png" rel="nofollow noreferrer">this</a></p>
  <p>but the output is the same list as the input. And then I tried to use <strong>SplitJson</strong> like <a href="https://i.stack.imgur.com/dkwFN.png" rel="nofollow noreferrer">this</a> but it doesn't work properly (it creates some GB of information inside the SplitJson processor)</p>
  <p>What am I doing wrong?</p>