Hi Flavio, Stephan was referring to
env.registerType(ExtendedClass1.class); env.registerType(ExtendedClass2.class); Cheers, Max On Wed, Feb 10, 2016 at 12:48 PM, Flavio Pompermaier <pomperma...@okkam.it> wrote: > What do you mean exactly..? Probably I'm missing something here..remember > that I can specify the right subClass only after the last flatMap, after the > first map neither me nor Flink can know the exact subclass of BaseClass > > On Wed, Feb 10, 2016 at 12:42 PM, Stephan Ewen <se...@apache.org> wrote: >> >> Class hierarchies should definitely work, even if the base class has no >> fields. >> >> They work more efficiently if you register the subclasses at the execution >> environment (Flink cannot infer them from the function signatures because >> the function signatures only contain the abstract base class). >> >> On Wed, Feb 10, 2016 at 12:23 PM, Flavio Pompermaier >> <pomperma...@okkam.it> wrote: >>> >>> Because The classes are not related to each other. Do you think it's a >>> good idea to have something like this? >>> >>> abstract class BaseClass(){ >>> String someField; >>> } >>> >>> class ExtendedClass1 extends BaseClass (){ >>> String someOtherField11; >>> String someOtherField12; >>> String someOtherField13; >>> ... >>> } >>> >>> class ExtendedClass2 extends BaseClass (){ >>> Integer someOtherField21; >>> Double someOtherField22; >>> Integer someOtherField23; >>> ... >>> } >>> >>> and then declare my map as Map<Tuple2,BaseClass>. and then apply a >>> flatMap that can be used to generated the specific datasets? >>> Doesn't this cause problem to Flink? Classes can be vrry different to >>> each other..maybe this can cause problems with the plan generation..isn't >>> it? >>> >>> Thanks Fabian and Stephan for the support! >>> >>> >>> On Wed, Feb 10, 2016 at 11:47 AM, Stephan Ewen <se...@apache.org> wrote: >>>> >>>> Why not use an abstract base class and N subclasses? >>>> >>>> On Wed, Feb 10, 2016 at 10:05 AM, Fabian Hueske <fhue...@gmail.com> >>>> wrote: >>>>> >>>>> Unfortunately, there is no Either<1,...,n>. >>>>> You could implement something like a Tuple3<Option<Type1>, >>>>> Option<Type2>, Option<Type3>>. However, Flink does not provide an Option >>>>> type (comes with Java8). You would need to implement it yourself incl. >>>>> TypeInfo and Serializer. You can get some inspiration from the Either type >>>>> info /serializer, if you want to go this way. >>>>> >>>>> Using a byte array would also work but doesn't look much easier than >>>>> the Option approach to me. >>>>> >>>>> 2016-02-10 9:47 GMT+01:00 Flavio Pompermaier <pomperma...@okkam.it>: >>>>>> >>>>>> Yes, the intermediate dataset I create then join again between >>>>>> themselves. What I'd need is a Either<1,...,n>. Is that possible to add? >>>>>> Otherwise I was thinking to generate a Tuple2<String,byte[]> and in >>>>>> the subsequent filter+map/flatMap deserialize only those elements I want >>>>>> to >>>>>> group togheter (e.g. t.f0=="someEventType") in order to generate the >>>>>> typed >>>>>> dataset based. >>>>>> Which one do you think is the best solution? >>>>>> >>>>>> On Wed, Feb 10, 2016 at 9:40 AM, Fabian Hueske <fhue...@gmail.com> >>>>>> wrote: >>>>>>> >>>>>>> Hi Flavio, >>>>>>> >>>>>>> I did not completely understand which objects should go where, but >>>>>>> here are some general guidelines: >>>>>>> >>>>>>> - early filtering is mostly a good idea (unless evaluating the filter >>>>>>> expression is very expensive) >>>>>>> - you can use a flatMap function to combine a map and a filter >>>>>>> - applying multiple functions on the same data set does not >>>>>>> necessarily materialize the data set (in memory or on disk). In most >>>>>>> cases >>>>>>> it prevents chaining, hence there is serialization overhead. In some >>>>>>> cases >>>>>>> where the forked data streams are joined again, the data set must be >>>>>>> materialized in order to avoid deadlocks. >>>>>>> - it is not possible to write a map that generates two different >>>>>>> types, but you could implement a mapper that returns an Either<First, >>>>>>> Second> type. >>>>>>> >>>>>>> Hope this helps, >>>>>>> Fabian >>>>>>> >>>>>>> 2016-02-10 8:43 GMT+01:00 Flavio Pompermaier <pomperma...@okkam.it>: >>>>>>>> >>>>>>>> Any help on this? >>>>>>>> >>>>>>>> On 9 Feb 2016 18:03, "Flavio Pompermaier" <pomperma...@okkam.it> >>>>>>>> wrote: >>>>>>>>> >>>>>>>>> Hi to all, >>>>>>>>> >>>>>>>>> in my program I have a Dataset that generated different types of >>>>>>>>> object wrt the incoming element. >>>>>>>>> Thus it's like a Map<Tuple2,Object>. >>>>>>>>> In order to type the different generated datasets I do something: >>>>>>>>> >>>>>>>>> Dataset<Tuple2> start =... >>>>>>>>> >>>>>>>>> Dataset<MyObj1> ds1 = start.filter().map(..); >>>>>>>>> Dataset<MyObj1> ds2 = start.filter().map(..); >>>>>>>>> Dataset<MyObj3> ds3 = start.filter().map(..); >>>>>>>>> Dataset<MyObj3> ds4 = start.filter().map(..); >>>>>>>>> >>>>>>>>> However this is very inefficient (I think because Flink needs to >>>>>>>>> materialize the entire source dataset for every slot). >>>>>>>>> >>>>>>>>> It's much more efficient to group the generation of objects of the >>>>>>>>> same type. E.g.: >>>>>>>>> >>>>>>>>> Dataset<Tuple2> start =.. >>>>>>>>> >>>>>>>>> Dataset<MyObj1> tmp1 = start.map(..); >>>>>>>>> Dataset<MyObj3> tmp2 = start.map(..); >>>>>>>>> Dataset<MyObj1> ds1 = tmp1.filter(); >>>>>>>>> Dataset<MyObj1> ds2 = tmp1.filter(); >>>>>>>>> Dataset<MyObj3> ds3 = tmp2.filter(); >>>>>>>>> Dataset<MyObj3> ds4 = tmp2.filter(); >>>>>>>>> >>>>>>>>> Increasing the number of slots per task manager make things worse >>>>>>>>> and worse :) >>>>>>>>> Is there a way to improve this situation? Is it possible to write a >>>>>>>>> "map" generating different type of object and then filter them by >>>>>>>>> generated >>>>>>>>> class type? >>>>>>>>> >>>>>>>>> Best, >>>>>>>>> Flavio >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>> >>>>>> >>>>>> >>>>> >>>> >>> >> >