能否在Apache NiFi中将JSON转换为固定宽度格式文件？

核心方案：使用ConvertRecord处理器 + 对应记录读写器 #

NiFi原生支持这类格式转换，无需自定义复杂脚本，通过配置ConvertRecord处理器搭配JSON记录读取器和固定宽度记录写入器即可完成需求。

1. 配置JSON记录读取器（解析输入JSON）

选择JsonTreeReader作为记录读取器，关键配置如下：

• Schema Access Strategy：选择Use 'Schema Text' Property
• Schema Text：定义匹配输入JSON结构的Schema，示例：

{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "orderName", "type": "string"}
  ]
}

该读取器会自动解析输入的JSON数组，将每个对象转换为一条独立记录。

2. 配置固定宽度记录写入器（生成目标格式）

选择FixedWidthRecordSetWriter作为记录写入器，这是实现固定宽度格式的核心组件，核心配置：

• Schema Access Strategy：选择Use 'Schema Text' Property
• Schema Text：定义每个字段的固定位置、宽度和对齐规则，示例（匹配你的预期输出）：

{
  "type": "record",
  "name": "FixedWidthOrder",
  "fields": [
    {
      "name": "orderId",
      "type": "string",
      "position": 0,
      "length": 10,
      "alignment": "left"
    },
    {
      "name": "orderName",
      "type": "string",
      "position": 11,
      "length": 8,
      "alignment": "left"
    }
  ]
}

参数说明：

• position：字段在每行的起始索引（从0开始计数）

• length：字段的固定宽度，内容长度不足时按alignment（此处为左对齐）补空格

• 两个字段之间的空格通过position控制：orderId占0-9位（共10个字符），orderName从11位开始，中间第10位即为分隔空格

• Line Ending Strategy：按需选择LF或CRLF，匹配目标文件的换行格式

3. 配置ConvertRecord处理器

将上述两个读写器关联到ConvertRecord处理器：

• Record Reader：选择你配置好的JsonTreeReader
• Record Writer：选择配置好的FixedWidthRecordSetWriter
• 其余参数保持默认即可，处理器会自动将输入的JSON数据流转换为固定宽度文本流

备选方案：使用ExecuteScript处理器自定义转换

如果需要更灵活的格式控制，也可以用ExecuteScript编写Groovy脚本实现，示例代码片段：

import groovy.json.JsonSlurper

def flowFile = session.get()
if (!flowFile) return

def slurper = new JsonSlurper()
def orders = slurper.parse(flowFile.read())

def output = new StringBuilder()
orders.each { order ->
    // 格式化orderId为10位左对齐，补空格
    def formattedId = String.format("%-10s", order.orderId)
    // 格式化orderName为8位左对齐，补空格
    def formattedName = String.format("%-8s", order.orderName)
    output.append(formattedId).append(" ").append(formattedName).append("\n")
}

flowFile = session.write(flowFile, { outputStream ->
    outputStream.write(output.toString().getBytes("UTF-8"))
} as OutputStreamCallback)

session.transfer(flowFile, REL_SUCCESS)

这种方式适合需要特殊格式逻辑的场景，但原生处理器方案更易维护和扩展。

内容的提问来源于stack exchange，提问作者janvit04