The returns(TypeInformation) and returns(String) methods do absolutely no type extraction, the user has to know what he is doing. If you read the methods description:
Pojo types such as org.my.MyPojo<myFieldName=TYPE0,myFieldName2=TYPE1> Generic types such as java.lang.ClassWith the returns(String) method you can create all types of type information we currently support.
returns(Class) the description is as follows:This method takes a class that will be analyzed by Flink's type extraction capabilities.
On 11.02.2015 21:42, Stephan Ewen wrote:
But in this case, there are no type parameters, correct? Centroid25 is not a generic class... On Wed, Feb 11, 2015 at 9:40 PM, Robert Metzger <rmetz...@apache.org> wrote:I think the issue is that the returns("eu.euranova.flink.Centroid25") variant only passes a string and the system does not know the typeparameters. So we have to put GenericTypeInfo there, because we basically see Object's. On Wed, Feb 11, 2015 at 9:37 PM, Stephan Ewen <se...@apache.org> wrote:@Timo If I understand it correctly, both omitting the "returns(...)" statement, or changing it to "returns(Centroid25.class)" would help? I think that the behavior between "returns(Centroid25.class)" and " returns("eu.euranova.flink.Centroid25")" should be consistent in thattheyboth handle the type as a POJO. Stephan On Wed, Feb 11, 2015 at 9:28 PM, Timo Walther <twal...@apache.org>wrote:Hey Nam-Luc, I think your problem lies in the following line: .returns("eu.euranova.flink.Centroid25") If you do not specify the fields of the class in the String by using "<myfield=String,otherField=int>", the underlying parser will create an "GenericTypeInfo" type information which then uses Kryo forserialization.In general, lambda expressions are a very new feature which currently makes a lot of problems due to missing type information by compilers.Maybeit is better to use (anonymous) classes instead. In case of "map()" functions you don't need to provide type hintsthroughthe "returns()" method. For other operators you need to either specify all fields of the classinthe String (makes no sense in you case) or you change the method to .returns(Centroid25.class) I hope that helps. Regards, Timo On 11.02.2015 17:38, Nam-Luc Tran wrote:Hello Stephan, Thank you for your help. I ensured all the POJO classes used comply to what you previously said and the same exception occurs. Here is the listing of classes Centroid25 and Point25: public class Centroid25 extends Point25 { public int id; public Centroid25() {} public Centroid25(int id, Double value0, Double value1, Double value2, Double value3, Double value4, Double value5, Double value6, Double value7, Double value8, Double value9, Double value10, Double value11, Double value12, Double value13, Double value14, Double value15, Double value16, Double value17, Double value18, Double value19, Double value20, Double value21, Double value22, Double value23, Double value24) { super(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20, value21, value22, value23, value24); this.id = id; } public Centroid25(int id, Point25 p) { super(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); this.id = id; } public Centroid25(int id, Tuple25 p) { super(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); this.id = id; } @Override public String toString() { return id + " " + super.toString(); } } public class Point25{ public Double f0,f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f14,f15,f16, f17,f18,f19,f20,f21,f22,f23,f24 = 0.0; public Point25() { } public Point25(Double value0, Double value1, Double value2, Double value3, Double value4, Double value5, Double value6, Double value7, Double value8, Double value9, Double value10, Double value11, Double value12, Double value13, Double value14, Double value15, Double value16, Double value17, Double value18, Double value19, Double value20, Double value21, Double value22, Double value23, Double value24) { f0=value0; f1=value1; f2=value2; f3=value3; f4=value4; f5=value5; f6=value6; f7=value7; f8=value8; f9=value9; f10=value10; f11=value11; f12=value12; f13=value13; f14=value14; f15=value15; f16=value16; f17=value17; f18=value18; f19=value19; f20=value20; f21=value21; f22=value22; f23=value23; f24=value24; } public List getFieldsAsList() { return Arrays.asList(f0, f1, f2, f3, f4, f5, f6, f7, f8, f9, f10, f11, f12, f13, f14, f15, f16, f17, f18, f19, f20, f21, f22, f23, f24); } public Point25 add(Point25 other) { f0 += other.f0; f1 += other.f1; f2 += other.f2; f3 += other.f3; f4 += other.f4; f5 += other.f5; f6 += other.f6; f7 += other.f7; f8 += other.f8; f9 += other.f9; f10 += other.f10; f11 += other.f11; f12 += other.f12; f13 += other.f13; f14 += other.f14; f15 += other.f15; f16 += other.f16; f17 += other.f17; f18 += other.f18; f19 += other.f19; f20 += other.f20; f21 += other.f21; f22 += other.f22; f23 += other.f23; f24 += other.f24; return this; } public Point25 div(long val) { f0 /= val; f1 /= val; f2 /= val; f3 /= val; f4 /= val; f5 += val; f6 += val; f7 += val; f8 += val; f9 += val; f10 += val; f11 += val; f12 += val; f13 += val; f14 += val; f15 += val; f16 += val; f17 += val; f18 += val; f19 += val; f20 += val; f21 += val; f22 += val; f23 += val; f24 += val; return this; } public double euclideanDistance(Point25 other) { List l = this.getFieldsAsList(); List ol = other.getFieldsAsList(); double res = 0; for(int i=0;iI came accross an error for which I am unable to retrace the exactcause.Starting from flink-java-examples module, I have extended the KMeans example to a case where points have 25 coordinates. It follows the exactsamestructure and transformations as the original example, only withpointshaving 25 coordinates instead of 2. When creating the centroids dataset within the code as follows thejobiterates and executes well: Centroid25 cent1 = newCentroid25(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 = newCentroid25(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 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.0with the following code: DataSet> 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 newCentroid25(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)atcom.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)atcom.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)atcom.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)atorg.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:205)atorg.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:210)at org.apache.flink.runtime.io.disk.InputViewIterator.next(InputViewIterator.java:43)atorg.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:91)at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496)atorg.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 statusFAILING.02/11/2015 14:58:27 CHAIN Map (Map atmain(DoTheKMeans.java:64)) ->Map (Map at main(DoTheKMeans.java:65))(1/1) switched to CANCELING 02/11/2015 14:58:27 Combine (Reduce atmain(DoTheKMeans.java:68))(1/1)switched to CANCELING 02/11/2015 14:58:27 CHAIN Reduce(Reduce atmain(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 toCANCELED 02/11/2015 14:58:27 Sync(BulkIteration (BulkIteration))(1/1) switchedto CANCELING 02/11/2015 14:58:27 Sync(BulkIteration (BulkIteration))(1/1) switchedto CANCELED 02/11/2015 14:58:27 CHAIN Map (Map atmain(DoTheKMeans.java:64)) ->Map (Map at main(DoTheKMeans.java:65))(1/1) switched to CANCELED 02/11/2015 14:58:27 Combine (Reduce atmain(DoTheKMeans.java:68))(1/1)switched to CANCELED 02/11/2015 14:58:27 CHAIN Reduce(Reduce atmain(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)atcom.esotericsoftware.kryo.io.Input.readVarInt(Input.java:355)at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:109)atcom.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)atcom.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)atorg.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:205)atorg.apache.flink.api.java.typeutils.runtime.KryoSerializer.deserialize(KryoSerializer.java:210)at org.apache.flink.runtime.io.disk.InputViewIterator.next(InputViewIterator.java:43)atorg.apache.flink.runtime.operators.NoOpDriver.run(NoOpDriver.java:91)at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496)atorg.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)atakka.actor.Actor$class.aroundReceive(Actor.scala:465)at org.apache.flink.runtime.client.JobClientListener.aroundReceive(JobClient.scala:74)atakka.actor.ActorCell.receiveMessage(ActorCell.scala:516)at akka.actor.ActorCell.invoke(ActorCell.scala:487) atakka.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 youhelp meretrace 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.htmlSent from the Apache Flink (Incubator) Mailing List archive. mailinglistarchive at Nabble.com.
