You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Deniz Koçak <le...@gmail.com> on 2022/02/01 20:01:22 UTC
GenericType problem in KeyedCoProcessFunction after disableGenericTypes()
Hi All,
We have a function extending `KeyedCoProcessFunction` and within that
function implementation. I wanted to keep a class object as a field
which is simply responsible for generating a UUID.
We disabled Kyro fallback for generic types via
`env.getConfig().disableGenericTypes()`.
I am receiving the error below when I pass an instance of
`UUIDGenerator` to the `KeyedCoProcessFunction` and try to run the
job.
***********************************************************************************
Exception in thread "main" java.lang.UnsupportedOperationException:
Generic types have been disabled in the ExecutionConfig and type
java.util.UUID is treated as a generic type.
***********************************************************************************
At that point I wonder how can/should I do to pass an instance of
`UUIDGenerator` to `PrebetModelRequestProcessor` function.
ModelRequestProcessor Class:
========================================================
public class ModelRequestProcessor extends
KeyedCoProcessFunction<String, CustomerInformation, CustomerRisk,
ModelRequest> {
protected final UUIDGenerator uuidGenerator;
public PrebetModelRequestProcessor(UUIDGenerator generator) {
this.uuidGenerator = generator;
}
}
========================================================
UUIDGenerator Class:
========================================================
public class UUIDGenerator implements Serializable {
private static final long serialVersionUID = 42L;
public java.util.UUID generate() {
return java.util.UUID.randomUUID();
}
}
========================================================
Re: GenericType problem in KeyedCoProcessFunction after disableGenericTypes()
Posted by Deniz Koçak <le...@gmail.com>.
Sorry for not finishing the last - 4th - question
4) When Kyro fallback is enabled, that means fallback is the last
resort if the object can be serialized with any other serializer like
`PojoSerializer`?
Can we still benefit from the performance of Flink built-in serializer
without sacrifying the flexibility of Kyro?
On Wed, Feb 2, 2022 at 11:22 AM Deniz Koçak <le...@gmail.com> wrote:
>
> Hi Yoel,
>
> Thank you for answering my question, really appreciated. However, that
> brings my mind more questions.
>
> 1) As far as I understood from the Flink built-in serialization, if a
> class is a POJO type wrt the items defined in POJOs section here [1],
> if an object is an instance of a POJO type class, no need to do
> anything extra. Serialization can be handled via `PojoSerializer`
> naively? In our case with `java.utils.UUID`, that class has no
> no-arguments public constructor which cause the fallback to Kyro I
> guess?
>
> 2) In your suggestion `you could provide a supplier of UUIDGenerator`,
> I am not sure what exactly you refer here? Something like a lambda
> which is capable of creating an instance of `UUIDGenerator` on demand,
> which can be called within open() method?
>
> 3) `java.util.UUID` class also implements `java.io.Serializable`, but
> as far as I understood that does not mean a class implementing
> `java.io.Serializable` is always be able to be serialized via Flink?
> At least with the built-in serializers (no referring Kyro)? UUID class
> is java Serializable, though still seems to be a generic type because
> does not fall into any of the categories mentioned in [1]?
>
> 4) When Kyro fallback is enabled, that means fallback is the last
> resort if the object can be serialized with
>
>
> [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/types_serialization/
>
> On Wed, Feb 2, 2022 at 6:27 AM Yoel Benharrous
> <yo...@gmail.com> wrote:
> >
> > Hi Deniz,
> >
> > You could declare UUIDGenerator as a transient field and instanciate it in the open
> > function
> >
> > Ot if you want to inject any UUIDGenerator you could provide a supplier of UUIDGenerator that should implement Serializable and invoke it in the open function.
> >
> >
> >
> >
> > On Tue, Feb 1, 2022, 10:01 PM Deniz Koçak <le...@gmail.com> wrote:
> >>
> >> Hi All,
> >>
> >> We have a function extending `KeyedCoProcessFunction` and within that
> >> function implementation. I wanted to keep a class object as a field
> >> which is simply responsible for generating a UUID.
> >>
> >> We disabled Kyro fallback for generic types via
> >> `env.getConfig().disableGenericTypes()`.
> >> I am receiving the error below when I pass an instance of
> >> `UUIDGenerator` to the `KeyedCoProcessFunction` and try to run the
> >> job.
> >>
> >> ***********************************************************************************
> >> Exception in thread "main" java.lang.UnsupportedOperationException:
> >> Generic types have been disabled in the ExecutionConfig and type
> >> java.util.UUID is treated as a generic type.
> >> ***********************************************************************************
> >>
> >> At that point I wonder how can/should I do to pass an instance of
> >> `UUIDGenerator` to `PrebetModelRequestProcessor` function.
> >>
> >>
> >> ModelRequestProcessor Class:
> >> ========================================================
> >> public class ModelRequestProcessor extends
> >> KeyedCoProcessFunction<String, CustomerInformation, CustomerRisk,
> >> ModelRequest> {
> >>
> >> protected final UUIDGenerator uuidGenerator;
> >>
> >> public PrebetModelRequestProcessor(UUIDGenerator generator) {
> >> this.uuidGenerator = generator;
> >> }
> >> }
> >> ========================================================
> >>
> >> UUIDGenerator Class:
> >> ========================================================
> >> public class UUIDGenerator implements Serializable {
> >>
> >> private static final long serialVersionUID = 42L;
> >>
> >> public java.util.UUID generate() {
> >> return java.util.UUID.randomUUID();
> >> }
> >> }
> >> ========================================================
Re: GenericType problem in KeyedCoProcessFunction after disableGenericTypes()
Posted by Deniz Koçak <le...@gmail.com>.
Hi Yoel,
Thank you for answering my question, really appreciated. However, that
brings my mind more questions.
1) As far as I understood from the Flink built-in serialization, if a
class is a POJO type wrt the items defined in POJOs section here [1],
if an object is an instance of a POJO type class, no need to do
anything extra. Serialization can be handled via `PojoSerializer`
naively? In our case with `java.utils.UUID`, that class has no
no-arguments public constructor which cause the fallback to Kyro I
guess?
2) In your suggestion `you could provide a supplier of UUIDGenerator`,
I am not sure what exactly you refer here? Something like a lambda
which is capable of creating an instance of `UUIDGenerator` on demand,
which can be called within open() method?
3) `java.util.UUID` class also implements `java.io.Serializable`, but
as far as I understood that does not mean a class implementing
`java.io.Serializable` is always be able to be serialized via Flink?
At least with the built-in serializers (no referring Kyro)? UUID class
is java Serializable, though still seems to be a generic type because
does not fall into any of the categories mentioned in [1]?
4) When Kyro fallback is enabled, that means fallback is the last
resort if the object can be serialized with
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/types_serialization/
On Wed, Feb 2, 2022 at 6:27 AM Yoel Benharrous
<yo...@gmail.com> wrote:
>
> Hi Deniz,
>
> You could declare UUIDGenerator as a transient field and instanciate it in the open
> function
>
> Ot if you want to inject any UUIDGenerator you could provide a supplier of UUIDGenerator that should implement Serializable and invoke it in the open function.
>
>
>
>
> On Tue, Feb 1, 2022, 10:01 PM Deniz Koçak <le...@gmail.com> wrote:
>>
>> Hi All,
>>
>> We have a function extending `KeyedCoProcessFunction` and within that
>> function implementation. I wanted to keep a class object as a field
>> which is simply responsible for generating a UUID.
>>
>> We disabled Kyro fallback for generic types via
>> `env.getConfig().disableGenericTypes()`.
>> I am receiving the error below when I pass an instance of
>> `UUIDGenerator` to the `KeyedCoProcessFunction` and try to run the
>> job.
>>
>> ***********************************************************************************
>> Exception in thread "main" java.lang.UnsupportedOperationException:
>> Generic types have been disabled in the ExecutionConfig and type
>> java.util.UUID is treated as a generic type.
>> ***********************************************************************************
>>
>> At that point I wonder how can/should I do to pass an instance of
>> `UUIDGenerator` to `PrebetModelRequestProcessor` function.
>>
>>
>> ModelRequestProcessor Class:
>> ========================================================
>> public class ModelRequestProcessor extends
>> KeyedCoProcessFunction<String, CustomerInformation, CustomerRisk,
>> ModelRequest> {
>>
>> protected final UUIDGenerator uuidGenerator;
>>
>> public PrebetModelRequestProcessor(UUIDGenerator generator) {
>> this.uuidGenerator = generator;
>> }
>> }
>> ========================================================
>>
>> UUIDGenerator Class:
>> ========================================================
>> public class UUIDGenerator implements Serializable {
>>
>> private static final long serialVersionUID = 42L;
>>
>> public java.util.UUID generate() {
>> return java.util.UUID.randomUUID();
>> }
>> }
>> ========================================================
Re: GenericType problem in KeyedCoProcessFunction after disableGenericTypes()
Posted by Yoel Benharrous <yo...@gmail.com>.
Hi Deniz,
You could declare UUIDGenerator as a transient field and instanciate it in
the open
function
Ot if you want to inject any UUIDGenerator you could provide a supplier of
UUIDGenerator that should implement Serializable and invoke it in the open
function.
On Tue, Feb 1, 2022, 10:01 PM Deniz Koçak <le...@gmail.com> wrote:
> Hi All,
>
> We have a function extending `KeyedCoProcessFunction` and within that
> function implementation. I wanted to keep a class object as a field
> which is simply responsible for generating a UUID.
>
> We disabled Kyro fallback for generic types via
> `env.getConfig().disableGenericTypes()`.
> I am receiving the error below when I pass an instance of
> `UUIDGenerator` to the `KeyedCoProcessFunction` and try to run the
> job.
>
>
> ***********************************************************************************
> Exception in thread "main" java.lang.UnsupportedOperationException:
> Generic types have been disabled in the ExecutionConfig and type
> java.util.UUID is treated as a generic type.
>
> ***********************************************************************************
>
> At that point I wonder how can/should I do to pass an instance of
> `UUIDGenerator` to `PrebetModelRequestProcessor` function.
>
>
> ModelRequestProcessor Class:
> ========================================================
> public class ModelRequestProcessor extends
> KeyedCoProcessFunction<String, CustomerInformation, CustomerRisk,
> ModelRequest> {
>
> protected final UUIDGenerator uuidGenerator;
>
> public PrebetModelRequestProcessor(UUIDGenerator generator) {
> this.uuidGenerator = generator;
> }
> }
> ========================================================
>
> UUIDGenerator Class:
> ========================================================
> public class UUIDGenerator implements Serializable {
>
> private static final long serialVersionUID = 42L;
>
> public java.util.UUID generate() {
> return java.util.UUID.randomUUID();
> }
> }
> ========================================================
>