提取某些json对象nifi Json

Question

我正在尝试提取 json 对象并将其存储到 hdfs。我的目标是 message 属性，即a6,b6,c6,d6,e6

json 示例

{
   "@timestamp":"2020-07-06T07:35:29.047Z",
   "@metadata":{
      "beat":"filebeat",
      "type":"_doc",
      "version":"7.7.1"
   },
   "log":{
      "offset":91,
      "file":{
         "path":"C:\\Program Files\\Filebeat\\test-kafka\\test_csv.csv"
      }
   },
   "message":"a6,b6,c6,d6,e6",
   "input":{
      "type":"log"
   },
   "ecs":{
      "version":"1.5.0"
   },
   "host":{
      "name":"host"
   },
   "agent":{
      "version":"7.7.1",
      "type":"filebeat",
      "ephemeral_id":"0b4a288f-f7ac-4db9-835e-60ca07a45fff",
      "hostname":"host",
      "id":"5e2fec03-bbdc-4f91-acc9-4ab36c7268db"
   }
}

GenerateFlowFile 属性

JsonEvaluatePath 属性

但是问题JsonEvaluatePath 没有按我的预期工作，我认为它只会提取消息属性。

hadoop@ambari:~$ hdfs dfs -cat /user/test/5a422f02-9074-4384-a3c9-f3e3ce7c2e40
{
   "@timestamp":"2020-07-06T07:35:29.047Z",
   "@metadata":{
      "beat":"filebeat",
      "type":"_doc",
      "version":"7.7.1"
   },
   "log":{
      "offset":91,
      "file":{
         "path":"C:\\Program Files\\Filebeat\\test-kafka\\test_csv.csv"
      }
   },
   "message":"a6,b6,c6,d6,e6",
   "input":{
      "type":"log"
   },
   "ecs":{
      "version":"1.5.0"
   },
   "host":{
      "name":"host"
   },
   "agent":{
      "version":"7.7.1",
      "type":"filebeat",
      "ephemeral_id":"0b4a288f-f7ac-4db9-835e-60ca07a45fff",
      "hostname":"host",
      "id":"5e2fec03-bbdc-4f91-acc9-4ab36c7268db"
   }
}

我错过了什么吗？

Answer 1

由于您使用 EvaluateJsonPath 并将目标设置为流文件属性，因此它将消息提取到流文件属性中，并且流文件的内容仍然与以前相同。您需要在 PutHDFS 之前使用另一个处理器（如 AttributesToJson）来使用您想要的属性重写流文件内容。另一种方法可能是将 EvaluateJsonPath 目标设置为流文件内容，但我不确定这是否会产生有效的 json。