我正在尝试在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”列都转储到配置单元表的一列中,并且该表将由“ partition_date”列进行分区,该列将成为表的第二列。问题是由于某种原因,NiFi在从“ Raw_Json”列中推断嵌套的json时遇到了问题,并且像表中的Null一样将其转储,如下所示:
有谁知道NiFi如何让NiFi将“ Raw_Json”列的整个嵌套Json读取为一列并将其发送到配置单元表?我该如何创建自己的avro模式?任何有关如何解决此问题的见解或想法将不胜感激!
通常,只要输入文件格式假定始终相同,就只需创建或生成(推断)avro模式一次-两个字段Raw_Json
和partition_date
。
例如,您应该在文件中包含以下内容avro-schema.json
:
{
"type" : "record",
"name" : "test",
"fields" : [ {
"name" : "Raw_Json",
"type" :
...
}, {
"name" : "partition_date",
"type" : "string",
"doc" : "Type inferred from '\"2017-05-19\"'"
} ]
}
并使用此文件作为Record Schema
在ConvertJSONToAvro
处理器。
列的类型Raw_Json
:
或者您必须使用所有嵌套字段,数组等完全定义复杂的数据类型。
或者,如果您要将Raw_Json
into string列的内容写入,则必须先将其转换为string,然后再将文件转换为avro。您可以使用的序列EvaluateJsonPath
和AttributesToJson
处理器。
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句