I feed Avro data as follows, turning it into SparseInstances, and leaving
Kryo serde for all internal streams in the topology.
bin/samoa samza target/SAMOA-Samza-0.4.0-incubating-SNAPSHOT.jar
"PrequentialEvaluation -e F1ClassificationPerformanceEvaluator -i -1 -l
(classifiers.ensemble.Bagging -s 10) -s (AvroFileStream -s
HDFSFileStreamSource -f /tmp/order_and_feats_flat_avro/2016_02_26/ -c 1 -e
binary -w 200:1) -f 10000"
It works fine in Local mode, but when deployed to Samza, I get the
following error. SparseInstanceData, which it complains it lacks a default
constructor, must have worked with Kryo SerDe at some point.
org.apache.samza.system.SystemConsumersException: Cannot deserialize an
incoming message for SystemStream [system=kafka0,
stream=Prequential_20160316182522-0-1-0]
at
org.apache.samza.system.SystemConsumers.update(SystemConsumers.scala:303)
at
org.apache.samza.system.SystemConsumers.tryUpdate(SystemConsumers.scala:270)
at org.apache.samza.system.SystemConsumers.org
$apache$samza$system$SystemConsumers$$poll(SystemConsumers.scala:258)
at
org.apache.samza.system.SystemConsumers$$anonfun$refresh$2.apply(SystemConsumers.scala:286)
at
org.apache.samza.system.SystemConsumers$$anonfun$refresh$2.apply(SystemConsumers.scala:286)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.MapLike$DefaultKeySet.foreach(MapLike.scala:174)
at
scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at
scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)
at scala.collection.SetLike$class.map(SetLike.scala:93)
at scala.collection.AbstractSet.map(Set.scala:47)
at
org.apache.samza.system.SystemConsumers.refresh(SystemConsumers.scala:286)
at
org.apache.samza.system.SystemConsumers.choose(SystemConsumers.scala:211)
at
org.apache.samza.container.RunLoop$$anonfun$process$1$$anonfun$1.apply(RunLoop.scala:108)
at
org.apache.samza.container.RunLoop$$anonfun$process$1$$anonfun$1.apply(RunLoop.scala:108)
at
org.apache.samza.util.TimerUtils$class.updateTimer(TimerUtils.scala:37)
at org.apache.samza.container.RunLoop.updateTimer(RunLoop.scala:35)
at
org.apache.samza.container.RunLoop$$anonfun$process$1.apply$mcVJ$sp(RunLoop.scala:107)
at
org.apache.samza.util.TimerUtils$class.updateTimerAndGetDuration(TimerUtils.scala:51)
at
org.apache.samza.container.RunLoop.updateTimerAndGetDuration(RunLoop.scala:35)
at org.apache.samza.container.RunLoop.process(RunLoop.scala:106)
at org.apache.samza.container.RunLoop.run(RunLoop.scala:74)
at
org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:558)
at
org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer.scala:93)
at
org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:67)
at
org.apache.samza.container.SamzaContainer.main(SamzaContainer.scala)
Caused by: com.esotericsoftware.kryo.KryoException: Class cannot be created
(missing no-arg constructor): org.apache.samoa.instances.SparseInstanceData
Serialization trace:
instanceData (org.apache.samoa.core.SerializableInstance)
instance (org.apache.samoa.learners.InstanceContent)
instanceContent (org.apache.samoa.learners.InstanceContentEvent)
at com.esotericsoftware.kryo.Kryo.newInstantiator(Kryo.java:1050)
at com.esotericsoftware.kryo.Kryo.newInstance(Kryo.java:1062)
at
com.esotericsoftware.kryo.serializers.FieldSerializer.create(FieldSerializer.java:228)
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:217)
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
at
com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:648)
at
com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:699)
at
com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:611)
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:729)
at
org.apache.samoa.utils.SamzaKryoSerdeFactory$SamzaKryoSerde.fromBytes(SamzaKryoSerdeFactory.java:147)
at
org.apache.samza.serializers.SerdeManager.fromBytes(SerdeManager.scala:115)
at
org.apache.samza.system.SystemConsumers.update(SystemConsumers.scala:300)
... 28 more
