flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Nam-Luc Tran <namluc.t...@euranova.eu>
Subject kryoException : Buffer underflow
Date Wed, 11 Feb 2015 14:02:17 GMT
Hello,

I came accross an error for which I am unable to retrace the exact cause.
Starting from flink-java-examples module, I have extended the KMeans example
to a case where points have 25 coordinates. It follows the exact same
structure and transformations as the original example, only with points
having 25 coordinates instead of 2.

When creating the centroids dataset within the code as follows the job
iterates and executes well:

Centroid25 cent1 = new Centroid25(ThreadLocalRandom.current().nextInt(0,
1000),
-10.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,-0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,10.0);
Centroid25 cent2 = new Centroid25(ThreadLocalRandom.current().nextInt(0,
1000),
-1.0,-1.0,1.0,1.0,-1.0,1.0,1.0,1.0,1.0,1.0,1.0,-1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0);
DataSet<Centroid25> centroids = env.fromCollection(Arrays.asList(cent1,
cent2));

When reading from a csv file containing the following:
-10.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,-0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,10.0
-1.0,-1.0,1.0,1.0,-1.0,1.0,1.0,1.0,1.0,1.0,1.0,-1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0

with the following code:
DataSet<Centroid25>> centroids = env
				.readCsvFile("file:///home/nltran/res3.csv")
				.fieldDelimiter(",")
				.includeFields("1111111111111111111111111")
				.types(Double.class, Double.class, Double.class, Double.class,
Double.class, Double.class,
						Double.class, Double.class, Double.class, Double.class, Double.class,
Double.class,
						Double.class, Double.class, Double.class, Double.class, Double.class,
Double.class,
						Double.class, Double.class, Double.class, Double.class, Double.class,
Double.class,
						Double.class).map(p -> {
					return new Centroid25(ThreadLocalRandom.current().nextInt(0, 1000),
p.f0,p.f1,p.f2,p.f3,p.f4,p.f5,p.f6,p.f7,p.f8,p.f9,p.f10,p.f11,p.f12,p.f13,p.f14,p.f15,p.f16,p.f17,p.f18,p.f19,p.f20,p.f21,p.f22,p.f23,p.f24);
				}).returns("eu.euranova.flink.Centroid25");


I hit the following exception:

02/11/2015 14:58:27	PartialSolution (BulkIteration (Bulk Iteration))(1/1)
switched to FAILED 
com.esotericsoftware.kryo.KryoException: Buffer underflow
	at
org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.require(NoFetchingInput.java:76)
	at com.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)
	at
com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
	at
org.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:205)
	at
org.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:210)
	at
org.apache.flink.runtime.io.disk.InputViewIterator.next(InputViewIterator.java:43)
	at org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:91)
	at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496)
	at
org.apache.flink.runtime.iterative.task.AbstractIterativePactTask.run(AbstractIterativePactTask.java:138)
	at
org.apache.flink.runtime.iterative.task.IterationHeadPactTask.run(IterationHeadPactTask.java:324)
	at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
	at
org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:204)
	at java.lang.Thread.run(Thread.java:745)

02/11/2015 14:58:27	Job execution switched to status FAILING.
02/11/2015 14:58:27	CHAIN Map (Map at main(DoTheKMeans.java:64)) -> Map (Map
at main(DoTheKMeans.java:65))(1/1) switched to CANCELING 
02/11/2015 14:58:27	Combine (Reduce at main(DoTheKMeans.java:68))(1/1)
switched to CANCELING 
02/11/2015 14:58:27	CHAIN Reduce(Reduce at main(DoTheKMeans.java:68)) -> Map
(Map at main(DoTheKMeans.java:71))(1/1) switched to CANCELING 
02/11/2015 14:58:27	DataSink(Print to System.out)(1/1) switched to CANCELED 
02/11/2015 14:58:27	Sync(BulkIteration (Bulk Iteration))(1/1) switched to
CANCELING 
02/11/2015 14:58:27	Sync(BulkIteration (Bulk Iteration))(1/1) switched to
CANCELED 
02/11/2015 14:58:27	CHAIN Map (Map at main(DoTheKMeans.java:64)) -> Map (Map
at main(DoTheKMeans.java:65))(1/1) switched to CANCELED 
02/11/2015 14:58:27	Combine (Reduce at main(DoTheKMeans.java:68))(1/1)
switched to CANCELED 
02/11/2015 14:58:27	CHAIN Reduce(Reduce at main(DoTheKMeans.java:68)) -> Map
(Map at main(DoTheKMeans.java:71))(1/1) switched to CANCELED 
02/11/2015 14:58:27	Job execution switched to status FAILED.
Exception in thread "main"
org.apache.flink.runtime.client.JobExecutionException:
com.esotericsoftware.kryo.KryoException: Buffer underflow
	at
org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.require(NoFetchingInput.java:76)
	at com.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)
	at
com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
	at
org.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:205)
	at
org.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:210)
	at
org.apache.flink.runtime.io.disk.InputViewIterator.next(InputViewIterator.java:43)
	at org.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:91)
	at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496)
	at
org.apache.flink.runtime.iterative.task.AbstractIterativePactTask.run(AbstractIterativePactTask.java:138)
	at
org.apache.flink.runtime.iterative.task.IterationHeadPactTask.run(IterationHeadPactTask.java:324)
	at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
	at
org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:204)
	at java.lang.Thread.run(Thread.java:745)

	at
org.apache.flink.runtime.client.JobClientListener$$anonfun$receiveWithLogMessages$2.applyOrElse(JobClient.scala:88)
	at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
	at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
	at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
	at
org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:37)
	at
org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:30)
	at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
	at
org.apache.flink.runtime.ActorLogMessages$$anon$1.applyOrElse(ActorLogMessages.scala:30)
	at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
	at
org.apache.flink.runtime.client.JobClientListener.aroundReceive(JobClient.scala:74)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
	at akka.actor.ActorCell.invoke(ActorCell.scala:487)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
	at akka.dispatch.Mailbox.run(Mailbox.scala:221)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
	at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

The centroid25 data is exactly the same in both cases. Could you help me
retrace what is wrong?

Thanks and best regards,

Tran Nam-Luc



--
View this message in context: http://apache-flink-incubator-mailing-list-archive.1008284.n3.nabble.com/kryoException-Buffer-underflow-tp3760.html
Sent from the Apache Flink (Incubator) Mailing List archive. mailing list archive at Nabble.com.

Mime
View raw message