NiFi 1.6.0如何转换S3数组JSON为换行分隔格式导入Redshift

前置说明 #

原始数组格式JSON直接导入Redshift会触发Invalid JSONPath format: Member is not an object.报错，根因是文件首尾包裹数组的[、]不符合Redshift对JSON加载的格式要求，必须转为每行一个独立JSON对象的换行分隔格式。
原始S3文件格式示例：

[
  {
    "a": 1,
    "b": 2
  },
  {
    "a": 3,
    "b": 4
  }
]

目标转换后格式：

{
  "a": 1,
  "b": 2
}
{
  "a": 3,
  "b": 4
}

本次待处理文件113MB、超1400万行，绝对不能用SplitJson、JoltTransformJSON这类会把全量文件加载到内存的处理器，这类方案要么直接触发OOM，要么拆出数百万小FlowFile打满NiFi内部存储，处理耗时极长。下面的纯流式文本替换方案内存占用稳定在几十MB级别，单文件处理仅需数秒，完全适配大文件场景。

完整流程链路 #

按顺序配置以下处理器即可：

• ListS3：配置目标S3桶的访问凭证、桶名、文件前缀，轮询待处理的JSON文件
• FetchS3Object：拉取ListS3命中的文件流内容，全程做流式读取不做全量内存加载
• ReplaceText（第一步：移除首尾数组符号）：核心配置如下
  • 替换策略：Regex Replace
  • 字符集：UTF-8
  • 搜索值：^\s*\[\s*|\s*\]\s*$
  • 替换值：留空，不填任何内容
  • 评估模式：Entire text
    该配置会流式匹配文件最开头的[（忽略前后空白、换行符）和文件最末尾的]（忽略前后空白、换行符），直接删除这两个数组包裹符号。
• ReplaceText（第二步：替换对象间分隔符）：核心配置如下
  • 替换策略：Regex Replace
  • 字符集：UTF-8
  • 搜索值：(\})\s*,\s*(\{)
  • 替换值：$1\n$2
  • 评估模式：Entire text
    该配置会自动匹配所有JSON对象之间的分隔结构（不管},和{中间夹了多少空格、换行、制表符），统一替换为}后接换行再跟{，输出标准的换行分隔JSON。
• （可选）ValidateRecord：配置JSON Reader，校验转换后的每行JSON格式合法性，自动路由坏数据到单独队列做排查
• 下游输出：可以直接把处理后的流Put回S3，再触发Redshift COPY命令加载，也可以直接对接Redshift相关处理器写入。

避坑提示 #

• 不要尝试用SplitJson拆分单个JSON对象：113MB文件会拆出数百万个小FlowFile，NiFi的FlowFile仓库和元数据管理会直接被压垮，处理耗时会从数秒拉长到数十分钟。
• 不要用JoltTransformJSON做格式转换：该处理器会把整个文件内容加载到JVM内存构建JSON树做转换，大文件下极易触发OOM。
• 转换后的换行分隔JSON直接用Redshift COPY命令的FORMAT JSON 'auto'参数加载即可，无需额外格式处理，加载性能最优。

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