【发布时间】：2019-03-04 10:42:23
【问题描述】：
我正在尝试使用 Apache Nifi 中的 JoltTransformRecord 将 json 转换为 json。当我尝试在https://jolt-demo.appspot.com/ 中转换 json 时，我得到了正确的结果。这没关系。
但是，当我尝试使用 JoltTransformRecord 转换 json 时，它会引发异常。错误是； “无法转换类 [Ljava.lang.Object; 因为该类型不受支持”的值”。但我不明白为什么我会收到这个错误。我确实检查了我的输入和输出模式，但我没有找到任何东西。它们看起来是正确的。
在下面，给出了我的输入和输出 json 示例、jolt 规范、输入和输出模式。另外，为此，我使用 JsonTreeReader 和 JsonRecordSetWriter。
--- 我该如何解决这个问题？ ---
example input json for JoltTransformRecord（在这个例子中，数组里面只有一个json对象，但实际上数组里面有很多json对象）；
[ {
"uuid" : "MFMS1-MC5",
"componentId" : "path1",
"Samples" : {
"PathFeedrate" : [ {
"dataItemId" : "pf",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68104",
"value" : "425.5333",
"name" : "Fact",
"subType" : "ACTUAL"
}, {
"dataItemId" : "pf",
"timestamp" : "2019-03-01T21:48:30.244219Z",
"sequence" : "68117",
"value" : "0",
"name" : "Fact",
"subType" : "ACTUAL"
} ]
},
"Events" : {
"SequenceNumber" : [ {
"dataItemId" : "seq",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68105",
"value" : "0",
"name" : "sequenceNum"
} ],
"Unit" : [ {
"dataItemId" : "unit",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68106",
"value" : "13",
"name" : "unitNum"
} ]
}
}]
我想要的示例输出 Json;
{
"DataItems" : [ {
"uuid" : "MFMS1-MC5",
"componentId" : "path1",
"eventType" : "Samples",
"type" : "PathFeedrate",
"dataItemId" : "pf",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68104",
"value" : "425.5333",
"name" : "Fact",
"subType" : "ACTUAL"
}, {
"uuid" : "MFMS1-MC5",
"componentId" : "path1",
"eventType" : "Samples",
"type" : "PathFeedrate",
"dataItemId" : "pf",
"timestamp" : "2019-03-01T21:48:30.244219Z",
"sequence" : "68117",
"value" : "0",
"name" : "Fact",
"subType" : "ACTUAL"
}, {
"uuid" : "MFMS1-MC5",
"componentId" : "path1",
"eventType" : "Events",
"type" : "SequenceNumber",
"dataItemId" : "seq",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68105",
"value" : "0",
"name" : "sequenceNum"
}, {
"uuid" : "MFMS1-MC5",
"componentId" : "path1",
"eventType" : "Events",
"type" : "Unit",
"dataItemId" : "unit",
"timestamp" : "2019-03-01T21:48:27.940558Z",
"sequence" : "68106",
"value" : "13",
"name" : "unitNum"
} ]
}
我的 Jolt 规范;
[
{
"operation": "shift",
"spec": {
"Samples": {
"*": {
"*": {
"@(3,uuid)": "Items.&2[#2].uuid",
"@(3,componentId)": "Items.&2[#2].componentId",
"$2": "Items.&2[#2].eventType",
"$1": "Items.&2[#2].type",
"*": "Items.&2[#2].&"
}
}
},
"Events": {
"*": {
"*": {
"@(3,uuid)": "Items.&2[#2].uuid",
"@(3,componentId)": "Items.&2[#2].componentId",
"$2": "Items.&2[#2].eventType",
"$1": "Items.&2[#2].type",
"*": "Items.&2[#2].&"
}
}
},
"Condition": {
"*": {
"*": {
"@(3,uuid)": "Items.&2[#2].uuid",
"@(3,componentId)": "Items.&2[#2].componentId",
"$2": "Items.&2[#2].eventType",
"$1": "Items.&2[#2].value",
"*": "Items.&2[#2].&"
}
}
}
}
},
{
"operation": "shift",
"spec": {
"Items": {
"*": {
"*": "DataItems[]"
}
}
}
}
]
此规范运行正常。因为我在 Jolt 变换演示中尝试过。
我正在使用 JsonTreeReader 来读取 JoltTransformRecord 中的 json。这是我的输入架构；
{
"name": "Items",
"namespace": "Items",
"type": "record",
"fields": [
{
"name": "uuid",
"type": "string"
},
{
"name": "componentId",
"type": "string"
},
{
"name": "Samples",
"type": ["null", {
"type": "map",
"values": {
"type": "array",
"items": {
"name": "SamplesDataItem",
"type": "record",
"fields": [
{
"name": "dataItemId",
"type": "string"
},
{
"name": "timestamp",
"type": "string"
},
{
"name": "sequence",
"type": "string"
},
{
"name": "value",
"type": "string"
},
{
"name": "name",
"type": ["null", "string"]
},
{
"name": "subType",
"type": ["null", "string"]
},
{
"name": "sampleRate",
"type": ["null", "string"]
},
{
"name": "statistic",
"type": ["null", "string"]
},
{
"name": "duration",
"type": ["null", "string"]
},
{
"name": "sampleCount",
"type": ["null", "string"]
},
{
"name": "compositionId",
"type": ["null", "string"]
},
{
"name": "resetTriggered",
"type": ["null", "string"]
}
]
}
}
}]
},
{
"name": "Events",
"type": ["null", {
"type": "map",
"values": {
"type": "array",
"items": {
"name": "EventsDataItem",
"type": "record",
"fields": [
{
"name": "dataItemId",
"type": "string"
},
{
"name": "timestamp",
"type": "string"
},
{
"name": "sequence",
"type": "string"
},
{
"name": "value",
"type": "string"
},
{
"name": "name",
"type": ["null", "string"]
},
{
"name": "subType",
"type": ["null", "string"]
},
{
"name": "compositionId",
"type": ["null", "string"]
},
{
"name": "resetTriggered",
"type": ["null", "string"]
}
]
}
}
}]
},
{
"name": "Condition",
"type": ["null", {
"type": "map",
"values": {
"type": "array",
"items": {
"name": "ConditionDataItem",
"type": "record",
"fields": [
{
"name": "dataItemId",
"type": "string"
},
{
"name": "timestamp",
"type": "string"
},
{
"name": "type",
"type": "string"
},
{
"name": "sequence",
"type": "string"
},
{
"name": "name",
"type": ["null", "string"]
},
{
"name": "subType",
"type": ["null", "string"]
},
{
"name": "nativeCode",
"type": ["null", "string"]
},
{
"name": "nativeSeverity",
"type": ["null", "string"]
},
{
"name": "qualifier",
"type": ["null", "string"]
},
{
"name": "statistic",
"type": ["null", "string"]
},
{
"name": "compositionId",
"type": ["null", "string"]
}
]
}
}
}]
}
]
}
我正在使用 JsonRecordSetWriter 将转换结果写入 JoltTransformRecord。这是我的输出模式；
{
"name": "Items",
"type": "record",
"namespace": "Items",
"fields": [
{
"name": "DataItems",
"type": {
"type": "array",
"items": {
"name": "DataItems",
"type": "record",
"fields": [
{
"name": "uuid",
"type": "string"
},
{
"name": "componentId",
"type": "string"
},
{
"name": "eventType",
"type": "string"
},
{
"name": "type",
"type": "string"
},
{
"name": "dataItemId",
"type": "string"
},
{
"name": "timestamp",
"type": "string"
},
{
"name": "value",
"type": "string"
},
{
"name": "name",
"type": ["null", "string"],
"default": null
},
{
"name": "subType",
"type": ["null", "string"],
"default": null
}
]
}
}
}
]
}
【问题讨论】：
-
由于你的输入输出是JSON，你试过原来的JoltTransformJson处理器吗？如果这也不起作用，那么 JOLT 库中可能存在错误，如果确实有效，则 JoltTransformRecord 处理器中可能存在错误。
-
JoltTransformJson 工作正常，没有任何错误。
标签： avro apache-nifi jolt