Storm 版本:1.2.1, Java版本:8
我正在用scala编写一个storm拓扑,并且在集群模式下运行它时开始出现以下错误。我能够在 LocalCluster
模式下以及配置:conf.put(Config.TOPOLOGY_TESTING_ALWAYS_TRY_SERIALIZE, Boolean.box( true))
获得相同的结果。以下是跟踪:
2018-05-05 00:49:59,342 ERROR util [Thread-37-disruptor-executor[6 6]-send-queue] Async loop died!
java.lang.RuntimeException: java.lang.RuntimeException: java.io.NotSerializableException: com.fasterxml.jackson.databind.node.ObjectNode
at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:522) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:487) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:74) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.disruptor$consume_loop_STAR_$fn__4492.invoke(disruptor.clj:84) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) [storm-core-1.2.1.jar:1.2.1]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131]
Caused by: java.lang.RuntimeException: java.io.NotSerializableException: com.fasterxml.jackson.databind.node.ObjectNode
at org.apache.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:41) ~[storm-core-1.2.1.jar:1.2.1]
at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:628) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:100) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:40) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:534) ~[kryo-3.0.3.jar:?]
at org.apache.storm.serialization.KryoValuesSerializer.serializeInto(KryoValuesSerializer.java:44) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.serialization.KryoTupleSerializer.serialize(KryoTupleSerializer.java:44) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.worker$assert_can_serialize.invoke(worker.clj:133) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.worker$mk_transfer_fn$fn__5204.invoke(worker.clj:213) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.executor$start_batch_transfer__GT_worker_handler_BANG_$fn__4882.invoke(executor.clj:314) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.disruptor$clojure_handler$reify__4475.onEvent(disruptor.clj:41) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:509) ~[storm-core-1.2.1.jar:1.2.1]
... 6 more
Caused by: java.io.NotSerializableException: com.fasterxml.jackson.databind.node.ObjectNode
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_131]
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) ~[?:1.8.0_131]
at org.apache.storm.serialization.SerializableSerializer.write(SerializableSerializer.java:38) ~[storm-core-1.2.1.jar:1.2.1]
at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:628) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:100) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.serializers.CollectionSerializer.write(CollectionSerializer.java:40) ~[kryo-3.0.3.jar:?]
at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:534) ~[kryo-3.0.3.jar:?]
at org.apache.storm.serialization.KryoValuesSerializer.serializeInto(KryoValuesSerializer.java:44) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.serialization.KryoTupleSerializer.serialize(KryoTupleSerializer.java:44) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.worker$assert_can_serialize.invoke(worker.clj:133) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.worker$mk_transfer_fn$fn__5204.invoke(worker.clj:213) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.daemon.executor$start_batch_transfer__GT_worker_handler_BANG_$fn__4882.invoke(executor.clj:314) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.disruptor$clojure_handler$reify__4475.onEvent(disruptor.clj:41) ~[storm-core-1.2.1.jar:1.2.1]
at org.apache.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:509) ~[storm-core-1.2.1.jar:1.2.1]
... 6 more
看来storm正在尝试序列化ObjectNode
,但它无法执行并给出NotSerializedException
。
ObjectNode
不应该可序列化
吗?我看到一个关于这个的旧讨论here ,但感觉这应该是可序列化
。
我尝试在 Storm 配置中添加以下内容,但没有帮助。
conf.registerSerialization(classOf[com.fasterxml.jackson.databind.node.ObjectNode])
我还尝试添加 conf.setSkipMissingKryoRegistrations(false)
,但再次没有救援。
什么是正确的解决方案?
最佳答案
从@Stig's answer获得灵感从此 answer ,每当在 bolt 之间传递该对象而不是我的对象时,我都会序列化该对象。所以现在我在 bolt 中发送这样的字节数组:
val messages = input.asInstanceOf[TupleImpl].get("Request").asInstanceOf[Array[Byte]].getObj[List[myObject]]
val objMapper = new ObjectMapper()
messages.foreach(message => collector.emit(new Values(objMapper.writeValueAsBytes(message))))
编辑1:
解决此问题的另一种可能的方法似乎(没有尝试,我通过发送字节解决)是为从一个 bolt 传递到另一个 bolt 的对象编写一个序列化器类,如所述 here 。以下是来自此链接的示例序列化程序:
public class StockAvroSerializer extends Serializer<Stock> {
private static final Logger LOG = LoggerFactory.getLogger(StockAvroSerializer.class);
private Schema SCHEMA = Stock.getClassSchema();
public void write(Kryo kryo, Output output, Stock object) {
DatumWriter<Stock> writer = new SpecificDatumWriter<>(SCHEMA);
ByteArrayOutputStream out = new ByteArrayOutputStream();
BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
try {
writer.write(object, encoder);
encoder.flush();
} catch (IOException e) {
LOG.error(e.toString(), e);
}
IOUtils.closeQuietly(out);
byte[] outBytes = out.toByteArray();
output.writeInt(outBytes.length, true);
output.write(outBytes);
}
public Stock read(Kryo kryo, Input input, Class<Stock> type) {
byte[] value = input.getBuffer();
SpecificDatumReader<Stock> reader = new SpecificDatumReader<>(SCHEMA);
Stock record = null;
try {
record = reader.read(null, DecoderFactory.get().binaryDecoder(value, null));
} catch (IOException e) {
LOG.error(e.toString(), e);
}
return record;
}
}
编辑2:
Here我发现为什么ObjectNode无法序列化:
JsonNode
does not know how to serialize itself with only information available at serialization: there is no ObjectMapper or JsonGenerator to use; latter is the component it must have to serialize itself (and contents if any). It can not and should try to instantiate either (how should they be configured?); and static singletons tend to cause problems in bigger systems (one part tries to go and configure them one way, another different)
但这是相当古老的通信,在新版本中,我相信应该有某种机制使其可序列化。
关于java - 在 Storm 拓扑中出现 NotSerializedException 错误,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50181889/