如何用NiFi将JSON对象插入PostgreSQL jsonb列？

针对CSV列名每日变化、需存入PostgreSQL jsonb列的场景，以下两种原生处理器组合方案可解决你遇到的字段映射、JSON语法错误问题：

方案一：ConvertRecord + PutDatabaseRecord（原生处理器，无需脚本） #

步骤1：配置CSVReader（适配动态列） #

在ConvertRecord的Record Reader中选择CSVReader，关键配置：

• Schema Access Strategy：选择Infer Schema（自动推断CSV列名，适配每日变化的字段）
• Infer Schema Headers：设置为First Line（用CSV第一行作为字段名）
• 其余配置保持默认，确保能正确读取任意列结构的CSV

步骤2：配置JsonRecordSetWriter（封装目标字段） #

在ConvertRecord的Record Writer中选择JsonRecordSetWriter，修改以下配置：

• Schema Access Strategy：选择Use 'Schema Text' Property
• Schema Text：定义输出结构，将整个CSV行的字段打包为json_content字段，示例：

  {
    "type": "record",
    "name": "OutputRecord",
    "fields": [
      {
        "name": "json_content",
        "type": "string"
      }
    ]
  }
  

• Record Path Value：为json_content字段赋值，填写${record:toJson()}（将当前Record对象转为纯JSON字符串）
• Write Mode：选择JSON Lines（每行输出一个独立的JSON对象）

步骤3：配置PutDatabaseRecord #

• Record Reader：选择JsonTreeReader（读取上一步输出的JSON Lines格式）
• Database Connection Pooling Service：配置你的PostgreSQL连接池
• Table Name：填写public.test_table
• Column Names：填写json_content（若ID为自增列，无需填写，由数据库自动生成）
• Field Names：填写json_content（与输出的JSON字段名对应）
• Data Type Mapping：确保json_content映射到PostgreSQL的jsonb类型（自动映射异常时手动指定）

方案二：FetchS3Object → ExecuteScript（Groovy）→ PutDatabaseRecord（灵活处理复杂场景） #

若方案一的record:toJson()仍出现MapRecord语法错误，可通过Groovy脚本直接处理CSV内容，规避Record对象序列化问题：

步骤1：ExecuteScript处理器配置 #

• Script Engine：选择Groovy
• Script Body：复制以下脚本，直接将CSV每行转为JSON字符串并封装为带json_content的JSON对象：

  import org.apache.commons.csv.CSVFormat
  import org.apache.commons.csv.CSVParser
  import groovy.json.JsonBuilder
  
  def flowFile = session.get()
  if (!flowFile) return
  
  flowFile = session.write(flowFile, { inputStream, outputStream ->
      def reader = new BufferedReader(new InputStreamReader(inputStream))
      def parser = CSVFormat.DEFAULT.withHeader().parse(reader)
      def writer = new BufferedWriter(new OutputStreamWriter(outputStream))
  
      parser.each { record ->
          def jsonMap = record.toMap()
          def jsonObj = new JsonBuilder([json_content: jsonMap]).toString()
          writer.write(jsonObj + "\n")
      }
  
      parser.close()
      reader.close()
      writer.close()
  } as StreamCallback)
  
  session.transfer(flowFile, REL_SUCCESS)
  

步骤2：PutDatabaseRecord配置 #

与方案一的步骤3一致，用JsonTreeReader读取输出的JSON Lines，映射json_content到表的jsonb列即可。

错误原因说明 #

1. 字段映射失败：之前ConvertRecord输出的是原始CSV字段（如ideez、name），但PutDatabaseRecord需要与表列匹配的json_content字段，字段名不匹配导致映射错误。
2. 缺少必填列：输出的记录中无json_content字段，PutDatabaseRecord找不到对应值抛出错误。
3. MapRecord语法错误：直接用JsonRecordSetWriter输出Record对象时，默认序列化会保留内部类型标识（如MapRecord），导致PostgreSQL无法识别为合法JSON，用record:toJson()或Groovy脚本生成纯JSON字符串即可解决。

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