A question about Kryo and Window State

[Vishal Santoshi]

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Vishal Santoshi]

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Vishal Santoshi]

Any more insight?

On Wed, Jun 13, 2018, 3:34 PM Vishal Santoshi <[hidden email]> wrote:

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Fabian Hueske-2]

Hi Vishal,

In general, Kryo serializers are not very upgrade friendly.

Serializer compatibility [1] might be right approach here, but Gordon (in CC) might know more about this.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/custom_serialization.html#handling-serializer-upgrades-and-compatibility

2018-06-18 12:06 GMT+02:00 Vishal Santoshi <[hidden email]>:

Any more insight?

On Wed, Jun 13, 2018, 3:34 PM Vishal Santoshi <[hidden email]> wrote:

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Tzu-Li (Gordon) Tai]

Hi Vishal,

Kryo has a serializer called `CompatibleFieldSerializer` that allows for simple backward compatibility changes, such as adding non-optional fields / removing fields.

If using the KryoSerializer is a must, then a good thing to do is to register Kryo's `CompatibleFieldSerializer` as the serializer to be used for that specific type.

By default, Kryo doesn’t use the `CompatibleFieldSerializer`, so you would have to explicitly register this.

The issue is, I think it wouldn’t be possible to use the `CompatibleFieldSerializer` _after_ the bytes were already written with the default, non-compatible version (the `FieldSerializer`).

So, AFAIK, this should only work if your Kryo state type was registered with the `CompatibleFieldSerializer` from the very beginning.

There could be a workaround, but unfortunately that would require changing some code in the `KryoSerializer`.

If you require this because your job is already running in production and data was already written by the `FieldSerializer`, please let me know and I’ll go into more details about this.

Cheers,

Gordon

On 21 June 2018 at 10:14:15 AM, Fabian Hueske ([hidden email]) wrote:

Hi Vishal,

In general, Kryo serializers are not very upgrade friendly.

Serializer compatibility [1] might be right approach here, but Gordon (in CC) might know more about this.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/custom_serialization.html#handling-serializer-upgrades-and-compatibility

2018-06-18 12:06 GMT+02:00 Vishal Santoshi <[hidden email]>:

Any more insight?

On Wed, Jun 13, 2018, 3:34 PM Vishal Santoshi <[hidden email]> wrote:

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Vishal Santoshi]

Thanks.

On Thu, Jun 21, 2018 at 4:34 AM, Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi Vishal,

Kryo has a serializer called `CompatibleFieldSerializer` that allows for simple backward compatibility changes, such as adding non-optional fields / removing fields.

If using the KryoSerializer is a must, then a good thing to do is to register Kryo's `CompatibleFieldSerializer` as the serializer to be used for that specific type.

By default, Kryo doesn’t use the `CompatibleFieldSerializer`, so you would have to explicitly register this.

The issue is, I think it wouldn’t be possible to use the `CompatibleFieldSerializer` _after_ the bytes were already written with the default, non-compatible version (the `FieldSerializer`).

So, AFAIK, this should only work if your Kryo state type was registered with the `CompatibleFieldSerializer` from the very beginning.

There could be a workaround, but unfortunately that would require changing some code in the `KryoSerializer`.

If you require this because your job is already running in production and data was already written by the `FieldSerializer`, please let me know and I’ll go into more details about this.

Cheers,

Gordon

On 21 June 2018 at 10:14:15 AM, Fabian Hueske ([hidden email]) wrote:

Hi Vishal,

In general, Kryo serializers are not very upgrade friendly.

Serializer compatibility [1] might be right approach here, but Gordon (in CC) might know more about this.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/custom_serialization.html#handling-serializer-upgrades-and-compatibility

2018-06-18 12:06 GMT+02:00 Vishal Santoshi <[hidden email]>:

Any more insight?

On Wed, Jun 13, 2018, 3:34 PM Vishal Santoshi <[hidden email]> wrote:

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.

[Vishal Santoshi]

Actually, yes. I have a job already running with "FieldSerializer" in production.  Any insights will be appreciated.

On Sat, Jun 23, 2018 at 7:39 AM, Vishal Santoshi <[hidden email]> wrote:

Thanks.

On Thu, Jun 21, 2018 at 4:34 AM, Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi Vishal,

Kryo has a serializer called `CompatibleFieldSerializer` that allows for simple backward compatibility changes, such as adding non-optional fields / removing fields.

If using the KryoSerializer is a must, then a good thing to do is to register Kryo's `CompatibleFieldSerializer` as the serializer to be used for that specific type.

By default, Kryo doesn’t use the `CompatibleFieldSerializer`, so you would have to explicitly register this.

The issue is, I think it wouldn’t be possible to use the `CompatibleFieldSerializer` _after_ the bytes were already written with the default, non-compatible version (the `FieldSerializer`).

So, AFAIK, this should only work if your Kryo state type was registered with the `CompatibleFieldSerializer` from the very beginning.

There could be a workaround, but unfortunately that would require changing some code in the `KryoSerializer`.

If you require this because your job is already running in production and data was already written by the `FieldSerializer`, please let me know and I’ll go into more details about this.

Cheers,

Gordon

On 21 June 2018 at 10:14:15 AM, Fabian Hueske ([hidden email]) wrote:

Hi Vishal,

In general, Kryo serializers are not very upgrade friendly.

Serializer compatibility [1] might be right approach here, but Gordon (in CC) might know more about this.

Best, Fabian

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/state/custom_serialization.html#handling-serializer-upgrades-and-compatibility

2018-06-18 12:06 GMT+02:00 Vishal Santoshi <[hidden email]>:

Any more insight?

On Wed, Jun 13, 2018, 3:34 PM Vishal Santoshi <[hidden email]> wrote:

Any ideas on the standard way ( or any roundabout way ) of doing a version upgrade that looks back ward compatible. 

The  @FieldSerializer.Optional("0") actually does  ignore the field ( even if reset ) giving it the default value if kyro is used. It has to do with the FieldSerializer behaves  .  There is another Serializer ( Composite I believe ) that allows for such back ward compatible changes.

I know some work is being done in 1.6 to allow for above use case and I think Google Data Flow does provide some avenues.

Thanks much

Vishal

On Tue, Jun 12, 2018 at 11:30 PM, Vishal Santoshi <[hidden email]> wrote:

I have a running pipe with Window State in a class say

Class A{

     long a;

}

It uses the default KryoSerializer 

I want to add a field to 

Class A {

  long a;

  long b;

}

I need to suspend with SP and resume with the new version of Class A

Is there a definite way to do this. I tried

Class A {

  long a;

   @FieldSerializer.Optional("0")

  long b;

}

but that seems to default to 0 , even when the Aggregation is putting in values.

Could somebody give pointers as to how to solve this 

Thanks a ton.