将 json 数组转换为 Nifi 中的 csv 文件

Question

InvokeHTTP 后我有复杂的 json 响应，需要使用 NiFi 将其转换为多个 csv 文件。例如下面的数据需要转换为三个 csv 文件。第一个 csv 文件将包含 ID、所有者和地址。第二个 csv 将有 ID、联系人、名字。第三个将有 ID、workList、所有者。

 "data" : [ {
    "ID":"789379832",
    "CreatedDate" : "2021-11-13T14:25:17-05:00",
    "Owner" : "abc",
    "workList" : "Sing, Play",
    "address" : "",
    "Phone" : "",
    "contact" : [ "Phone","Email" ],
    "email" : "",
    "firstName" : "Kim",
    "notes" : "Hello abc",
    "yearCreated" : 2021
  }, {
    "ID":"999079832",
    "CreatedDate" : "2021-04-13T09:36:51-04:00",
     "Owner" : "xyz",
    "workList" : "Sing",
    "address" : "80 Hastings Street",
    "Phone" : "",
    "contact" : [ "Phone","Self" ],
    "email" : "",
    "firstName" : "Hari",
    "notes" : "Hello Hari",
    "yearCreated" : 2021
  }
  ]

示例 csv1

ID,Owner,address
789379832,abc,
999079832,xyz,80 Hastings Street

我正在使用“EvaluateJsonPath”通过 $.data 获取关键“数据”。然后使用“ReplaceText”和“SplitJson”数组使 json 数组成为文件流内容。然后我使用 'JsonTreeReader' 和 'CsvRecordSetWriter' 的“ConvertRecord”处理器。我想了解如何在此处为 ConvertRecord 中的每个 csv 文件设置属性值。我的方法不是最优的，我应该改用 JoltTransformJSON 吗？

Answer 1

编辑：

好的，所以你想制作 3 个 CSV 文件，其中每个数组项都是 CSV 中的一行。

与我最初的建议非常相似，但使用 EvaluateJsonPath 而不是 ForkRecord。

您不需要为此流中的任何一个使用 FlowFile 属性。

将目标属性设置为流文件内容的 EvaluateJsonPath。这避免了对 ReplaceText 的需要。
3 ConvertRecords 并将 EvaluateJsonPath 成功关系纳入所有 3 个（每个人都获得一份记录副本，用于您的 3 个 CSV）。每个 UpdateRecord 都有一个不同的 CSV 编写器，具有简化的架构以删除您不需要的 CSV 字段。然后把UpdateRecords成功关系放到PutFile中。

使用仅包含所需字段的架构设置每个 CSVRecordSetWriter，例如

{
  "type": "record",
  "name": "CSV",
  "fields" : [
    {"name": "ID", "type": "string"},
    {"name": "Owner", "type": "string"},
    {"name": "address", "type": "string"}
  ]
}

以便它忽略所有您不需要的字段。

Docs for avro schemas

Tool for validating avro schemas

带有数组的示例 Avro 架构（用于联系人字段）

{
  "type": "record",
  "name": "CSV",
  "fields" : [
    {"name": "ID", "type": "string"},
    {"name": "contact", "type":{"type": "array", "items" : "string" }},
    {"name": "firstName", "type": "string"}
  ]
}

原文：

所以首先，当您这么说时 For example below data needs to be converted to three csv files - 您的意思是数组中的每个对象都需要创建 3 个 CSV 文件吗？ IE。示例数据将为数组项 1 生成 3 个 CSV，为数组项 2 生成 3 个 CSV？

在这种情况下，我认为 ForkRecord is a better fit for you, as you can achive most of what you're doing in a single processor - you can process the elements of the array as individual records (see the Additional Details 页的示例）。使用 JSON Reader 和 CSV 编写器，您已经处理了转换和拆分。

之后，您可以放下 3 个 ConvertRecords，并将 ForkRecord 成功关系放入所有 3 个（每个人都获得一份记录副本，用于您的 3 个 CSV）。每个 UpdateRecord 都有一个不同的 CSV 编写器，具有简化的架构以删除您不需要的 CSV 字段。然后把UpdateRecords成功关系放到PutFile中。

将 json 数组转换为 Nifi 中的 csv 文件

convert json array to csv files in Nifi

csv

apache-nifi