如何在Apache NiFi中使用JoltTransformJson实现预期输出

云原生阿岚

2026-7-29

使用Apache NiFi的JoltTransformJson实现JSON格式转换

输入JSON

{"kind":"bresponse","schema":{"fields":[{"name":"export_time","type":"STRING","mode":"NULLABLE"},{"name":"account_id","type":"STRING","mode":"NULLABLE"},{"name":"cost","type":"FLOAT","mode":"NULLABLE"}]},"jobReference":{"projectId":"avach","jobId":"job_G4hQs","location":"ui"},"totalRows":"49","pageToken":"BG123","rows":[{"f":[{"v":"20230"},{"v":"0107F6-7"},{"v":"0.0"}]},{"f":[{"v":"20230101"},{"v":"01DC85-B"},{"v":"3.48E-4"}]}],"totalBytesProcessed":"37472","jobComplete":true,"cacheHit":false}

预期输出JSON

[{"export_time":"20230","account_id":"0107F6-7","cost":"0.0"},{"export_time":"20230101","account_id":"01DC85-B","cost":"3.48E-4"}]

Jolt转换规范

使用以下三级Shift操作的Jolt Spec即可完成转换：

[
  {
    "operation": "shift",
    "spec": {
      "schema": {
        "fields": {
          "*": {
            "name": "keys[&1]"
          }
        }
      },
      "rows": {
        "*": {
          "f": {
            "*": {
              "v": "values[&3][&1]"
            }
          }
        }
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "values": {
        "*": {
          "*": {
            "@": "&2.@(3,keys[&1])"
          }
        }
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": "[]"
    }
  }
]

转换步骤说明：

第一步Shift：提取schema.fields中的字段名称存入keys数组，同时将rows中每行的f数组内的v值提取到二维数组values，结构为values[行索引][字段索引]。
第二步Shift：将values中的值与keys对应的字段名关联，生成包含键值对的行对象。
第三步Shift：将所有行对象提取为顶层数组，得到最终预期格式。