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 that they
> both 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 for
> serialization.
> >
> > In general, lambda expressions are a very new feature which currently
> > makes a lot of problems due to missing type information by compilers.
> Maybe
> > it is better to use (anonymous) classes instead.
> >
> > In case of "map()" functions you don't need to provide type hints through
> > the "returns()" method.
> >
> > For other operators you need to either specify all fields of the class in
> > the 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;i
> >>
> >>> 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 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> 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.
> >>>
> >>>
> >>
> >>
> >
>
