如何在NiFi流中将完整JSON记录插入数据库JSON列？

场景与初始尝试 #

我通过NiFi从数据库A加载JSON格式记录，目标是将这些记录插入数据库B某表的JSON列，流程框架如下：

A => ExecuteSQLRecord => [jsonRow,...] => [?] => B.table(data json)

尝试使用JoltTransformJson+PutDatabaseRecord组合，流程如下：

[jsonRow,...] => JoltTransformJson => [{data:jsonRow},...] => PutDatabaseRecord => B.Table(data json)

但执行失败，报错信息：

Batch entry 0 insert into schema.table(data) values(('MapRecord [{...}]')) was aborted: ERROR: invalid input syntax for type json Detail: Token "MapRecord" is invalid. Where: JSON data, line 1: MapRecord... unnamed portal parameter $1 = '...' Call getNextException to see other errors in the batch.

测试验证 #

[2024-08-03] 搭建测试环境验证方案可行性，流程为：

GenerateFlowFile => JoltTransformJson => PutDatabaseRecord

数据流转路径：

[{id,name},{id,nam}] => [{data:{id,name}},{data:{id,name}}] => pgTable(id serial,data json)

PostgreSQL成功写入两行数据，证明JoltTransformJson+PutDatabaseRecord的组合方案是可行的，但生产环境中该流程仍报错，优先计划深入排查生产环境问题根源。

问题定位 #

[2024-08-03] 进一步测试发现Null值是问题诱因：当原始JSON记录包含Null值属性时，数据在发送到数据库前会被自动包裹为MapRecord[(...)]格式，导致PostgreSQL解析JSON时出错。需求明确：必须保留Null值（字段缺失测试是数据库数据验证的一部分），同时需要去除MapRecord的包裹。

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