从 CSV 转换为 JSON 时 Apache NiFi ConvertRecord 重命名字段

Posted 2023-03-31

【中文标题】从 CSV 转换为 JSON 时 Apache NiFi ConvertRecord 重命名字段

【英文标题】：Apache NiFi ConvertRecord Rename Fields While Converting from CSV to JSON

【发布时间】：2019-06-20 07:26:52

【问题描述】：

我在 CSV 文件中输入了以下示例数据：

Identifer Key,Name,Address,City,State,ZIP  
WELD-424,Jane Doe,123 Main St,Whereverville,CA,90210  
MOWN-175,John Doe,555 Broadway Ave,New York,NY,10010

目前我想出的处理器流程是：

获取文件 UpdateAttribute 使用架构文本设置 avro.schema 属性 PutMongoRecord 使用CSVReader 将记录加载到数据库中

Avro 架构会是什么样子？这是我最好的猜测（基于我关心的两个领域）：

   "type" : "record",
   "namespace" : "TheNameSpace",
   "name" : "MySchema",
   "fields" : [
      "name" : "Identifier Key" , "type" : ["string"]
      "name" : "Name" , "type" : ["string", "null"]
   ]

在上面指定"Identifier Key" 会出错，因为它包含一个空格。但是，"Name" 等其他字段可以正常加载。

我面临的一些挑战：

如何重命名字段？这是否需要在ConvertRecord 处理器和模式生态系统之外的另一个处理器块中完成？这似乎是一种常见情况，因为您希望字段具有来自许多不同来源的相同名称。 Avro 不喜欢其中包含空格的字段名称（因此从 "Identifier Key" -> "_id" 开始会有问题）。 似乎没有办法在读写操作期间重命名字段。我认为别名功能会有所帮助（例如：来自"Name" -> "fullName"） 在导入 MongoDB 之前将单个字段（即标识符键）全部小写？

我还尝试使用ConvertRecord 处理器块首先从 CSV 转换为 JSON，以便可以将其作为 JSON 导入 MongoDB。它需要看起来像这样（标识符键字段全部小写），但在ConvertRecord 运行后，标识符键的字段出现null：

"_id": "weld-424", "fullName": "Jane Doe", "updated": "$date":"2018-11-01T04:00:00.000Z", "created": "$date":"2018-11-01T04:00:00.000Z"
"_id": "mown-175", "fullName": "John Doe", "updated": "$date":"2018-11-01T04:00:00.000Z", "created": "$date":"2018-11-01T04:00:00.000Z"

【问题讨论】：

所以看起来我肯定需要一种方法来重命名字段，使它们不包含空格（为了 Avro 友好并重命名为数据导入的正确字段名称。

【参考方案1】：

关于什么 avro 我鼓励您阅读 avro specs 不是一个大文档，它将解释如何使用 Avro。 关于您的问题：

如何重命名字段？你可以使用Jolt transform processor。 Avro 不喜欢包含空格的字段名称： 您至少有 3 个选项： 忽略标题字段，因此将使用架构字段名称，注意您需要删除标题行并且字段必须以相同的顺序出现。 使用 Jolt。 在使用架构之前更改名称，您可以创建自己的处理器！

您的其他问题的答案与简历相同，如果我是您，我将创建不带空格的架构字段名称，然后将其更改为：

从标题中删除空格，like here，这样字段名称会抱怨 Avro 规范（如果它不以数字开头）。 使用 Jolt 将字段名称转换为数据库名称。 将您的数据放入数据库中。

希望对你有帮助。

【参考方案2】：

要禁用名称验证，我们需要定义 avro 模式注册表，Jira NiFI-4612 解决了这个问题。

验证字段名称

false

一旦您在AvroSchemaRegistry 中定义 avro 架构，我们就可以在 avro 架构中使用空格。

---

要更改字段名称，请使用 QueryRecord 处理器和 Record Reader/Writer（使用新别名） 控制器服务。

在QueryRecord处理器中添加新属性

select "Identifer Key" _id,Name,Address from FLOWFILE

QueryRecord 处理器的输出将有 _id,Name,Address 作为新字段名称。