需要帮助推断 NiFi 中 json 文件的 avro 模式

Question

我正在尝试在 NiFi 中创建一个流，该流采用有效的 json 文件并使用 PutHiveStreaming 处理器将其直接放入配置单元表中。我的 json 如下所示：

{
"Raw_Json": {
    "SystemInfo": {
        "Id": "a string ID",
        "TM": null,
        "CountID": "a string ID",
        "Topic": null,
        "AccountID": "some number",
        "StationID": "some number",
        "STime": "some Timestamp",
        "ETime": "some Timestamp"
    },
    "Profile": {
        "ID": "ID number",
        "ProductID": "Some Number",
        "City": "City Name",
        "State": "State Name",
        "Number": "XXX-XXX-XXXX",
        "ExtNumber": null,
        "Unit": null,
        "Name": "Person Name",
        "Service": "Purchase",
        "AddrID": "00000000",
        "Products": {
            "Product": [{
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"

            },
            {
                "Code": "CODE",
                "Description": "some description"
            }]
        }
    },
    "Total": {
        "Amount": "some amount",
        "Delivery": "some address",
        "Estimate": "some amount",
        "Tax": null,
        "Delivery_Type": null

    }

},
"partition_date":"2017-05-19"

}

我正在使用 InferAvroSchema 处理器获取 json，然后使用推断的 avro 模式将 json 转换为 avro 格式并将其发送到 PutHiveStreaming 处理器。我的流程看起来像这样：

主要目标是我希望将所有“Raw_Json”列转储到 hive 表中的一列中，并且该表将由“partition_date”列进行分区，该列将是表的第二列。问题是，由于某种原因，NiFi 在从“Raw_Json”列推断嵌套 json 时遇到问题，并将其像 Null 一样倾倒在桌子上，如下所示：

有谁知道如何让 NiFi 将“Raw_Json”列的整个嵌套 Json 作为一列读取并将其发送到 hive 表？我如何创建自己的 avro 架构来做到这一点？任何有关如何解决此问题的见解或想法将不胜感激！

Answer 1

通常，只要您的输入文件格式假定始终相同 - 两个字段 Raw_Json 和 partition_date，您只需创建或生成（推断）avro 架构一次。

你应该在文件中有这样的东西，例如avro-schema.json:

{
  "type" : "record",
  "name" : "test",
  "fields" : [ {
    "name" : "Raw_Json",
    "type" : 
    ...
  }, {
    "name" : "partition_date",
    "type" : "string",
    "doc" : "Type inferred from '\"2017-05-19\"'"
  } ]
}

并在ConvertJSONToAvro 处理器中将此文件用作Record Schema。

列的类型Raw_Json：

或者您必须使用所有嵌套字段、数组等完全定义复杂数据类型。

或者，如果您想将Raw_Json 的内容写入字符串列，那么您必须在将文件转换为 avro 之前将其转换为字符串。您可以使用EvaluateJsonPath 和AttributesToJson 处理器的序列。