带有 Morphlines 的 Avro 中的子记录

Question

我正在尝试使用 kite-sdk morphline 模块将 JSON 转换为 Avro。在玩了之后，我能够使用简单的模式（没有复杂的数据类型）将 JSON 转换为 Avro。

然后我更进一步，修改了 Avro 架构，如下所示 (subrec.avsc)。如您所见，架构由一个子记录组成。

当我尝试使用 morphlines.conf 和 subrec.avsc 将 JSON 转换为 Avro 时，它失败了。

不知何故 JSON 路径 "/record_type[]/alert/action" 没有被 toAvro 函数翻译。

morphlines.conf

morphlines : [
   {
   id : morphline1
   importCommands : ["org.kitesdk.**"]

   commands : [
      # Read the JSON blob
      { readJson: {} }

      { logError { format : "record: {}", args : ["@{}"] } }

      # Extract JSON
      { extractJsonPaths { flatten: false, paths: {
              "/record_type[]/alert/action" : /alert/action,
              "/record_type[]/alert/signature_id" : /alert/signature_id,
              "/record_type[]/alert/signature" : /alert/signature,
              "/record_type[]/alert/category" : /alert/category,
              "/record_type[]/alert/severity" : /alert/severity
      } } }

      { logError { format : "EXTRACTED THIS : {}", args : ["@{}"] } }

      { extractJsonPaths { flatten: false, paths: {
              timestamp : /timestamp,
              event_type : /event_type,
              source_ip : /src_ip,
              source_port : /src_port,
              destination_ip : /dest_ip,
              destination_port : /dest_port,
              protocol : /proto,
      } } }

      # Create Avro according to schema
      { logError { format : "WE GO TO AVRO"} }

      { toAvro { schemaFile : /etc/flume/conf/conf.empty/subrec.avsc } }

      # Create Avro container
      { logError { format : "WE GO TO BINARY"} }
      { writeAvroToByteArray { format: containerlessBinary } }

      { logError { format : "DONE!!!"} }
   ]
   }
]

还有subrec.avsc

{
  "type" : "record",
  "name" : "Event",
  "fields" : [ {
    "name" : "timestamp",
    "type" : "string"
  }, {
    "name" : "event_type",
    "type" : "string"
  }, {
    "name" : "source_ip",
    "type" : "string"
  }, {
    "name" : "source_port",
    "type" : "int"
  }, {
    "name" : "destination_ip",
    "type" : "string"
  }, {
    "name" : "destination_port",
    "type" : "int"
  }, {
    "name" : "protocol",
    "type" : "string"
  }, {
    "name": "record_type",
    "type" : ["null", {
      "name" : "alert",
      "type" : "record",
      "fields" : [ {
            "name" : "action",
            "type" : "string"
        }, {
            "name" : "signature_id",
            "type" : "int"
        }, {
            "name" : "signature",
            "type" : "string"
        }, {
            "name" : "category",
            "type" : "string"
        }, {
            "name" : "severity",
            "type" : "int"
        }
      ] } ]
  } ]
}

{ logError { format : "EXTRACTED THIS : {}", args : ["@{}"] } } 上的输出我输出如下：

[{
    /record_type[]/alert / action = [allowed], 
    /record_type[]/alert / category = [],
    /record_type[]/alert / severity = [3],
    /record_type[]/alert / signature = [GeoIP from NL,
    Netherlands],
    /record_type[]/alert / signature_id = [88006],
    _attachment_body = [{
            "timestamp": "2015-03-23T07:42:01.303046",
            "event_type": "alert",
            "src_ip": "1.1.1.1",
            "src_port": 18192,
            "dest_ip": "46.231.41.166",
            "dest_port": 62004,
            "proto": "TCP",
            "alert": {
                "action": "allowed",
                "gid": "1",
                "signature_id": "88006",
                "rev": "1",
                "signature" : "GeoIP from NL, Netherlands ",
                "category" : ""
                "severity" : "3"
                }
            }], 
    _attachment_mimetype=[json/java + memory],
    basename = [simple_eve.json]
}]

Answer 1

2017 年 6 月 22 日更新

您必须使用 addValues 或 setValues 填充结构中的数据才能使其正常工作

{
    addValues {
        micDefaultHeader : [
            {
                eventTimestampString : "2017-06-22 18:18:36"
            }
        ]
    }
}

---

在调试 morphline toAvro 的源之后，无论您在映射结构中放入什么，似乎记录都是要评估的第一个对象。

解决方案很简单，但不幸的是花了一点额外的时间，eclipse，在调试模式下运行 Flume 代理，克隆源代码和大量咖啡。

到此为止。

我的架构：

{
  "type" : "record",
  "name" : "co_lowbalance_event",
  "namespace" : "co.tigo.billing.cboss.lowBalance",
  "fields" : [ {
    "name" : "dummyValue",
    "type" : "string",
    "default" : "dummy"
  }, {
    "name" : "micDefaultHeader",
    "type" : {
      "type" : "record",
      "name" : "mic_default_header_v_1_0",
      "namespace" : "com.millicom.schemas.root.struct",
      "doc" : "standard millicom header definition",
      "fields" : [ {
        "name" : "eventTimestampString",
        "type" : "string",
        "default" : "12345678910"
      } ]
    }
  } ]
}

吗啉文件：

morphlines : [
        {
                id : convertJsonToAvro
                importCommands : ["org.kitesdk.**"]
                commands : [
                        {
                                readJson {
                                        outputClass : java.util.Map
                                }
                        }

                        {
                                addValues {
                                   micDefaultHeader : [{}]
                                }
                        }


                        {
                                logDebug { format : "my record: {}", args : ["@{}"] } 
                        }


                        {
                                toAvro {
                                        schemaFile : /home/asarubbi/Development/test/co_lowbalance_event.avsc
                                        mappings : {
                                                "micDefaultHeader" : micDefaultHeader
                                                "micDefaultHeader/eventTimestampString" : eventTimestampString
                                        }

                                }
                        }


                        {
                                writeAvroToByteArray {
                                        format : containerlessJSON
                                        codec : null
                                }
                        }
                ]
        }
]

魔法就在这里：

{
   addValues {
      micDefaultHeader : [{}]
   }
}

在映射中：

mappings : {
    "micDefaultHeader" : micDefaultHeader
    "micDefaultHeader/eventTimestampString" : eventTimestampString
}

解释：

在代码中，评估的第一个字段名称是 RECORD 类型的 micDefaultHeader。由于无法为 RECORD 指定默认值（逻辑上正确），因此 toAvro 代码对此进行评估，没有在映射中配置任何值，因此它在检测到（错误地）记录应该为空时失败吨。

但是，查看代码，您可能会发现它需要一个 Map 对象，不包含任何值来取悦解析器并继续下一个元素。

所以我们使用 addValues 添加一个地图对象，并用一个空地图 [{}] 填充它。请注意，这必须与导致您为空值的记录的名称相匹配。在我的情况下“micDefaultHeader”

如果您有更好的解决方案，请随时发表评论，因为这看起来像是一个“肮脏的修复”