如何在NiFi中使用Jolt Transform规范实现credits数组映射？

输入JSON #

{
  "schema": {
    "fields": [
      {
        "name": "user",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "billed",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "currency",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "currency_conversion_rate",
        "type": "FLOAT",
        "mode": "NULLABLE"
      },
      {
        "name": "usage_unit",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "adjustment_info_mode",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "adjustment_info_type",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "InvoiceMonth",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "credits",
        "type": "RECORD",
        "mode": "REPEATED",
        "fields": [
          {
            "name": "name",
            "type": "STRING",
            "mode": "NULLABLE"
          },
          {
            "name": "amount",
            "type": "FLOAT",
            "mode": "NULLABLE"
          },
          {
            "name": "full_name",
            "type": "STRING",
            "mode": "NULLABLE"
          },
          {
            "name": "id",
            "type": "STRING",
            "mode": "NULLABLE"
          },
          {
            "name": "type",
            "type": "STRING",
            "mode": "NULLABLE"
          }
        ]
      },
      {
        "name": "resource_name",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "resource_global_name",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "label",
        "type": "STRING",
        "mode": "NULLABLE"
      }
    ]
  },
  "InvoiceMonth": "202006",
  "credits": [
    {
      "v": {
        "f": [
          {
            "v": "ae391_credit:0"
          },
          {
            "v": "-0.010237"
          },
          {
            "v": "Committed dollar "
          },
          {
            "v": "ae31_credit:0"
          },
          {
            "v": "COMMITTED_BASE"
          }
        ]
      }
    },
    {
      "v": {
        "f": [
          {
            "v": "ae391_credit:0"
          },
          {
            "v": "-0.010237"
          },
          {
            "v": "Committed dollar "
          },
          {
            "v": "ae31_credit:0"
          },
          {
            "v": "COMMITTED_BASE"
          }
        ]
      }
    }
  ],
  "resource_name": "projects-0",
  "resource_global_name": "//compute.googleapis.com/",
  "label": ""
}

尝试的Jolt规范 #

[
  {
    "operation": "shift",
    "spec": {
      "credits": {
        "*": {
          "*": {
            "*": {
              "*": {
                "@v": "[&3].@(5,schema.fields.fields[&].name)"
              }
            }
          }
        }
      }
    }
  }
]

期望输出 #

[
  {
    "name": "ae391_credit:0",
    "amount": "-0.010237",
    "full_name": "Committed dollar",
    "id": "ae31_credit:0",
    "type": "COMMITTED_BASE"
  },
  {
    "name": "ae391_credit:0",
    "amount": "-0.010237",
    "full_name": "Committed dollar",
    "id": "ae31_credit:0",
    "type": "COMMITTED_BASE"
  }
]

正确的Jolt Transform规范 #

[
  {
    "operation": "shift",
    "spec": {
      "credits": {
        "*": {
          "v": {
            "f": {
              "*": {
                "@v": "[&3].@(6,schema.fields[8].fields[&].name)"
              }
            }
          }
        }
      }
    }
  },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "*": {
        "full_name": "=trim"
      }
    }
  }
]

规范说明 #

1. Shift操作：
   • 路径credits.*.v.f.*精准定位到每个credit条目的字段值，通过@v提取实际内容。
   • [&3]指定将值放入对应索引的数组元素中，&3是从当前节点向上回溯3层的索引，对应外层credits数组的每个元素位置。
   • @(6,schema.fields[8].fields[&].name)：向上回溯6层到根节点，找到schema里fields数组的第8个元素（类型为RECORD的credits字段），再取它的fields数组中对应索引的name作为键名，完成映射。
2. Modify操作：
   • 使用=trim函数去除full_name字段的首尾空格，和期望输出格式保持一致。

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