Apache Nifi：使用 UpdateRecord 处理器解析数据

Posted 2023-02-16

【中文标题】Apache Nifi：使用 UpdateRecord 处理器解析数据

【英文标题】：Apache Nifi: Parse data with UpdateRecord Processor

【发布时间】：2018-12-07 07:46:10

【问题描述】：

我正在尝试使用 UpdateRecord 处理器解析 Nifi (1.7.1) 中的一些数据。 原始数据是 json 文件，我想根据模式将其转换为 Avro。 Avro 转换没问题，但在该转换中，我还需要将 json 数据中的一个数组元素解析为 Avro 中的不同结构。 这是输入 json 的示例数据：

  "geometry" : 
"coordinates" : [ [ 4.963087975800593, 45.76365595859971 ], [ 4.962874487781098, 45.76320922779652 ], [ 4.962815443439148, 45.763116079159374 ], [ 4.962744732112515, 45.763010484202866 ], [ 4.962096825239138, 45.762112721939246 ] ]  ...

作为其架构（在 RecordReader 中指定）：

  "type": "record",
  "name": "features",
  "fields": [
    
      "name": "geometry",
      "type": 
        "type": "record",
        "name": "geometry",
        "fields": [
          
            "name": "coordinatesJson",
            "type": 
              "type": "array",
              "items": 
                "type": "array",
                "items": "double"
              
            
          ,
        ]
      
    ,
    ....
  ]
 

如您所见，坐标是一个数组数组。

我需要根据这个架构（在 RecordWriter 中指定）将这些数据解析到 Avro：

  "name": "outputdata",
  "type": "record",
  "fields": [
    "name": "coordinatesAvro",
      "type": 
        "type": "array",
        "items" : 
        "type" : "record",
        "name" : "coordinatesAvro",
        "fields" : [ 
          "name" : "X",
          "type" : "double"
        , 
          "name" : "Y",
          "type" : "double"
         ]
      
      
    ,
    .....

  ]
   

这里的问题是我无法使用 RecordPath 函数从坐标Json 解析到坐标Avro 我尝试了几种映射，例如：

Property:                            Value:
/coordinatesJson[0..-1]/X            /geometry/coordinatesAvro[*][0]
/coordinatesJson[0..-1]/Y            /geometry/coordinatesAvro[*][1]

这应该是一个非常简单的解析步骤，但正如我所说，我一直在绕圈子来实现这一目标。

任何帮助将不胜感激。

【参考方案1】：

当我遇到类似的事情时，我接下来会做： 1）通过ExecuteScript处理器将Json转换为具有我需要的结构的Json（例如在您的情况下：coordinateAvro）。我使用了 ECMAScript，因为您可以简单地解析 JSON 并使用对象（转换它们）。 2）ConvertJsonToAvro 具有一个用于 Reader 和 Writer 的通用模式（在您的情况下为 coordinatesAvro）。 它工作得很好，我已经在 BigData 案例中使用过它。这是您的问题的可能解决方案之一。