You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by 李 琳 <le...@outlook.com> on 2025/02/08 06:40:28 UTC

Skip Restore Incompatibility of Key Serializer for Savepoint

Dear All,

In a Flink job using statementSet, I have two DMLs for index calculation. I made a modification to one of the DMLs, changing the group keys from "distinct_id, $city, $province, window_start, window_end" to "distinct_id, window_start, window_end". When attempting to restore the job from a savepoint, Flink throws an exception due to an incompatibility in the key serializer between the original and the new state.

I'm seeking a way to make Flink skip the state related to the incompatible key serializer and proceed with restoring the unmodified DMLs. Are there any more effective approaches to achieve this?

Best regards, leilinee

RE: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by Ariana Allison <Al...@ecovetaevents.live>.
YES
I HELPED ,

HI
Thanks for response .

If you are interested to see the attendee list, I will send it to you .
Are you interested ?
I’m waiting for your response.
Thank you.

From: Gabor Somogyi <ga...@gmail.com>
Sent: Monday, March 17, 2025 11:14 AM
To: 李 琳 <le...@outlook.com>
Cc: user@flink.apache.org
Subject: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Good to hear it helped!

G


On Mon, Mar 17, 2025 at 3:50 AM 李 琳 <le...@outlook.com>> wrote:
Hi Gabor,

Thanks for your advice. I attempted to use the state processor API to resolve my issue, and it seems to be performing well.

First, I read the savepoint file and removed the state that couldn't be restored by the operator ID. Finally, I used the API to rewrite the new savepoint file, replacing the old one, the new Job can restor state and running well.

Best regards, leilinee

________________________________
发件人: Gabor Somogyi <ga...@gmail.com>>
发送时间: 2025年2月10日 16:48
收件人: 李 琳 <le...@outlook.com>>
抄送: user@flink.apache.org<ma...@flink.apache.org> <us...@flink.apache.org>>
主题: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Seems like I was not clear in my previous suggestions.
`execution.savepoint.ignore-unclaimed-state=true` would help only if FLINK-20375 [1] would be implemented.

Please has a look at the state processor API [2]. I would try something like:

SavepointWriter

    .fromExistingSavepoint(env, oldPath, new ...StateBackend())

    .removeOperator(OperatorIdentifier.forUid("your_operator_uid_which_is_problematic"))

    .write(newPath);
Hope this helps.

[1] https://issues.apache.org/jira/browse/FLINK-20375
[2] https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/

BR,
G


On Mon, Feb 10, 2025 at 4:27 AM 李 琳 <le...@outlook.com>> wrote:
Dear Gabor,

I sincerely appreciate your suggestion to set the configuration 'execution.state-recovery.ignore-unclaimed-state: true' in the Flink configuration file. I put it into my test.
However, unfortunately, it seems that this configuration didn't have the expected effect. When I ran my Flink job, it once again threw an exception, due to an incompatibility in the key serializer between the original and the new state. Given that the outcome didn't match our anticipation, I was wondering if you could share more in - depth details about this configuration.
Your insights would be of great help to me in resolving this issue. Thank you very much in advance.
Best regards，Leilinee

获取 Outlook for iOS<https://aka.ms/o0ukef>
________________________________
发件人: Gabor Somogyi <ga...@gmail.com>>
发送时间: Saturday, February 8, 2025 4:58:48 PM
收件人: 李 琳 <le...@outlook.com>>
抄送: user@flink.apache.org<ma...@flink.apache.org> <us...@flink.apache.org>>
主题: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Hi leilinee,

Since Flink SQL is not supporting operator uid configuration there are only hard ways.
State operator uid change + `execution.savepoint.ignore-unclaimed-state=true` would be the easy way.

Depending on you constraints either you use the state processor API and remove the
operator which belongs to this statement or you not using the savepoint at all (no savepoint restore).
Be aware that these ways you're going to loose state data so be careful.

If you have the possibility to invest more time then with the state processor API you
can do the state migration where you don't loose anything.

BR,
G


On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com>> wrote:
Dear All,

In a Flink job using statementSet, I have two DMLs for index calculation. I made a modification to one of the DMLs, changing the group keys from "distinct_id, $city, $province, window_start, window_end" to "distinct_id, window_start, window_end". When attempting to restore the job from a savepoint, Flink throws an exception due to an incompatibility in the key serializer between the original and the new state.

I'm seeking a way to make Flink skip the state related to the incompatible key serializer and proceed with restoring the unmodified DMLs. Are there any more effective approaches to achieve this?

Best regards, leilinee

Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by Gabor Somogyi <ga...@gmail.com>.
Good to hear it helped!

G


On Mon, Mar 17, 2025 at 3:50 AM 李 琳 <le...@outlook.com> wrote:

> Hi Gabor,
>
> Thanks for your advice. I attempted to use the state processor API to
> resolve my issue, and it seems to be performing well.
>
> First, I read the savepoint file and removed the state that couldn't be
> restored by the operator ID. Finally, I used the API to rewrite the new
> savepoint file, replacing the old one, the new Job can restor state and
> running well.
>
> Best regards, leilinee
>
> ------------------------------
> *发件人:* Gabor Somogyi <ga...@gmail.com>
> *发送时间:* 2025年2月10日 16:48
> *收件人:* 李 琳 <le...@outlook.com>
> *抄送:* user@flink.apache.org <us...@flink.apache.org>
> *主题:* Re: Skip Restore Incompatibility of Key Serializer for Savepoint
>
> Seems like I was not clear in my previous suggestions.
> `execution.savepoint.ignore-unclaimed-state=true` would help only if
> FLINK-20375 [1] would be implemented.
>
> Please has a look at the state processor API [2]. I would try something
> like:
>
> SavepointWriter    .fromExistingSavepoint(env, oldPath, new ...StateBackend())    .removeOperator(OperatorIdentifier.forUid("your_operator_uid_which_is_problematic"))    .write(newPath);
>
> Hope this helps.
>
> [1] https://issues.apache.org/jira/browse/FLINK-20375
> [2]
> https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/
>
> BR,
> G
>
>
> On Mon, Feb 10, 2025 at 4:27 AM 李 琳 <le...@outlook.com> wrote:
>
> Dear Gabor,
>
> I sincerely appreciate your suggestion to set the configuration
> 'execution.state-recovery.ignore-unclaimed-state: true' in the Flink
> configuration file. I put it into my test.
> However, unfortunately, it seems that this configuration didn't have the
> expected effect. When I ran my Flink job, it once again threw an exception,
> due to an incompatibility in the key serializer between the original and
> the new state. Given that the outcome didn't match our anticipation, I was
> wondering if you could share more in - depth details about this
> configuration.
> Your insights would be of great help to me in resolving this issue. Thank
> you very much in advance.
> Best regards，Leilinee
>
> 获取 Outlook for iOS <https://aka.ms/o0ukef>
> ------------------------------
> *发件人:* Gabor Somogyi <ga...@gmail.com>
> *发送时间:* Saturday, February 8, 2025 4:58:48 PM
> *收件人:* 李 琳 <le...@outlook.com>
> *抄送:* user@flink.apache.org <us...@flink.apache.org>
> *主题:* Re: Skip Restore Incompatibility of Key Serializer for Savepoint
>
> Hi leilinee,
>
> Since Flink SQL is not supporting operator uid configuration there are
> only hard ways.
> State operator uid change +
> `execution.savepoint.ignore-unclaimed-state=true` would be the easy way.
>
> Depending on you constraints either you use the state processor API and
> remove the
> operator which belongs to this statement or you not using the savepoint at
> all (no savepoint restore).
> Be aware that these ways you're going to loose state data so be careful.
>
> If you have the possibility to invest more time then with the state
> processor API you
> can do the state migration where you don't loose anything.
>
> BR,
> G
>
>
> On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com> wrote:
>
> Dear All,
>
> In a Flink job using statementSet, I have two DMLs for index calculation.
> I made a modification to one of the DMLs, changing the group keys from "*distinct_id,
> $city, $province, window_start, window_end*" to "*distinct_id,
> window_start, window_end*". When attempting to restore the job from a
> savepoint, Flink throws an exception due to an incompatibility in the key
> serializer between the original and the new state.
>
> I'm seeking a way to make Flink skip the state related to the incompatible
> key serializer and proceed with restoring the unmodified DMLs. Are there
> any more effective approaches to achieve this?
>
> Best regards, leilinee
>
>

回复: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by 李 琳 <le...@outlook.com>.
Hi Gabor,

Thanks for your advice. I attempted to use the state processor API to resolve my issue, and it seems to be performing well.

First, I read the savepoint file and removed the state that couldn't be restored by the operator ID. Finally, I used the API to rewrite the new savepoint file, replacing the old one, the new Job can restor state and running well.

Best regards, leilinee

________________________________
发件人: Gabor Somogyi <ga...@gmail.com>
发送时间: 2025年2月10日 16:48
收件人: 李 琳 <le...@outlook.com>
抄送: user@flink.apache.org <us...@flink.apache.org>
主题: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Seems like I was not clear in my previous suggestions.
`execution.savepoint.ignore-unclaimed-state=true` would help only if FLINK-20375 [1] would be implemented.

Please has a look at the state processor API [2]. I would try something like:

SavepointWriter
    .fromExistingSavepoint(env, oldPath, new ...StateBackend())
    .removeOperator(OperatorIdentifier.forUid("your_operator_uid_which_is_problematic"))
    .write(newPath);

Hope this helps.

[1] https://issues.apache.org/jira/browse/FLINK-20375
[2] https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/

BR,
G


On Mon, Feb 10, 2025 at 4:27 AM 李 琳 <le...@outlook.com>> wrote:
Dear Gabor,

I sincerely appreciate your suggestion to set the configuration 'execution.state-recovery.ignore-unclaimed-state: true' in the Flink configuration file. I put it into my test.
However, unfortunately, it seems that this configuration didn't have the expected effect. When I ran my Flink job, it once again threw an exception, due to an incompatibility in the key serializer between the original and the new state. Given that the outcome didn't match our anticipation, I was wondering if you could share more in - depth details about this configuration.
Your insights would be of great help to me in resolving this issue. Thank you very much in advance.
Best regards，Leilinee

获取 Outlook for iOS<https://aka.ms/o0ukef>
________________________________
发件人: Gabor Somogyi <ga...@gmail.com>>
发送时间: Saturday, February 8, 2025 4:58:48 PM
收件人: 李 琳 <le...@outlook.com>>
抄送: user@flink.apache.org<ma...@flink.apache.org> <us...@flink.apache.org>>
主题: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Hi leilinee,

Since Flink SQL is not supporting operator uid configuration there are only hard ways.
State operator uid change + `execution.savepoint.ignore-unclaimed-state=true` would be the easy way.

Depending on you constraints either you use the state processor API and remove the
operator which belongs to this statement or you not using the savepoint at all (no savepoint restore).
Be aware that these ways you're going to loose state data so be careful.

If you have the possibility to invest more time then with the state processor API you
can do the state migration where you don't loose anything.

BR,
G


On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com>> wrote:
Dear All,

In a Flink job using statementSet, I have two DMLs for index calculation. I made a modification to one of the DMLs, changing the group keys from "distinct_id, $city, $province, window_start, window_end" to "distinct_id, window_start, window_end". When attempting to restore the job from a savepoint, Flink throws an exception due to an incompatibility in the key serializer between the original and the new state.

I'm seeking a way to make Flink skip the state related to the incompatible key serializer and proceed with restoring the unmodified DMLs. Are there any more effective approaches to achieve this?

Best regards, leilinee

Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by Gabor Somogyi <ga...@gmail.com>.
Seems like I was not clear in my previous suggestions.
`execution.savepoint.ignore-unclaimed-state=true` would help only if
FLINK-20375 [1] would be implemented.

Please has a look at the state processor API [2]. I would try something
like:

SavepointWriter    .fromExistingSavepoint(env, oldPath, new
...StateBackend())
.removeOperator(OperatorIdentifier.forUid("your_operator_uid_which_is_problematic"))
   .write(newPath);

Hope this helps.

[1] https://issues.apache.org/jira/browse/FLINK-20375
[2]
https://nightlies.apache.org/flink/flink-docs-master/docs/libs/state_processor_api/

BR,
G


On Mon, Feb 10, 2025 at 4:27 AM 李 琳 <le...@outlook.com> wrote:

> Dear Gabor,
>
> I sincerely appreciate your suggestion to set the configuration
> 'execution.state-recovery.ignore-unclaimed-state: true' in the Flink
> configuration file. I put it into my test.
> However, unfortunately, it seems that this configuration didn't have the
> expected effect. When I ran my Flink job, it once again threw an exception,
> due to an incompatibility in the key serializer between the original and
> the new state. Given that the outcome didn't match our anticipation, I was
> wondering if you could share more in - depth details about this
> configuration.
> Your insights would be of great help to me in resolving this issue. Thank
> you very much in advance.
> Best regards，Leilinee
>
> 获取 Outlook for iOS <https://aka.ms/o0ukef>
> ------------------------------
> *发件人:* Gabor Somogyi <ga...@gmail.com>
> *发送时间:* Saturday, February 8, 2025 4:58:48 PM
> *收件人:* 李 琳 <le...@outlook.com>
> *抄送:* user@flink.apache.org <us...@flink.apache.org>
> *主题:* Re: Skip Restore Incompatibility of Key Serializer for Savepoint
>
> Hi leilinee,
>
> Since Flink SQL is not supporting operator uid configuration there are
> only hard ways.
> State operator uid change +
> `execution.savepoint.ignore-unclaimed-state=true` would be the easy way.
>
> Depending on you constraints either you use the state processor API and
> remove the
> operator which belongs to this statement or you not using the savepoint at
> all (no savepoint restore).
> Be aware that these ways you're going to loose state data so be careful.
>
> If you have the possibility to invest more time then with the state
> processor API you
> can do the state migration where you don't loose anything.
>
> BR,
> G
>
>
> On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com> wrote:
>
> Dear All,
>
> In a Flink job using statementSet, I have two DMLs for index calculation.
> I made a modification to one of the DMLs, changing the group keys from "*distinct_id,
> $city, $province, window_start, window_end*" to "*distinct_id,
> window_start, window_end*". When attempting to restore the job from a
> savepoint, Flink throws an exception due to an incompatibility in the key
> serializer between the original and the new state.
>
> I'm seeking a way to make Flink skip the state related to the incompatible
> key serializer and proceed with restoring the unmodified DMLs. Are there
> any more effective approaches to achieve this?
>
> Best regards, leilinee
>
>

Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by 李 琳 <le...@outlook.com>.
Dear Gabor,

I sincerely appreciate your suggestion to set the configuration 'execution.state-recovery.ignore-unclaimed-state: true' in the Flink configuration file. I put it into my test.
However, unfortunately, it seems that this configuration didn't have the expected effect. When I ran my Flink job, it once again threw an exception, due to an incompatibility in the key serializer between the original and the new state. Given that the outcome didn't match our anticipation, I was wondering if you could share more in - depth details about this configuration.
Your insights would be of great help to me in resolving this issue. Thank you very much in advance.
Best regards，Leilinee

获取 Outlook for iOS<https://aka.ms/o0ukef>
________________________________
发件人: Gabor Somogyi <ga...@gmail.com>
发送时间: Saturday, February 8, 2025 4:58:48 PM
收件人: 李 琳 <le...@outlook.com>
抄送: user@flink.apache.org <us...@flink.apache.org>
主题: Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Hi leilinee,

Since Flink SQL is not supporting operator uid configuration there are only hard ways.
State operator uid change + `execution.savepoint.ignore-unclaimed-state=true` would be the easy way.

Depending on you constraints either you use the state processor API and remove the
operator which belongs to this statement or you not using the savepoint at all (no savepoint restore).
Be aware that these ways you're going to loose state data so be careful.

If you have the possibility to invest more time then with the state processor API you
can do the state migration where you don't loose anything.

BR,
G


On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com>> wrote:
Dear All,

In a Flink job using statementSet, I have two DMLs for index calculation. I made a modification to one of the DMLs, changing the group keys from "distinct_id, $city, $province, window_start, window_end" to "distinct_id, window_start, window_end". When attempting to restore the job from a savepoint, Flink throws an exception due to an incompatibility in the key serializer between the original and the new state.

I'm seeking a way to make Flink skip the state related to the incompatible key serializer and proceed with restoring the unmodified DMLs. Are there any more effective approaches to achieve this?

Best regards, leilinee

Re: Skip Restore Incompatibility of Key Serializer for Savepoint

Posted by Gabor Somogyi <ga...@gmail.com>.
Hi leilinee,

Since Flink SQL is not supporting operator uid configuration there are only
hard ways.
State operator uid change +
`execution.savepoint.ignore-unclaimed-state=true` would be the easy way.

Depending on you constraints either you use the state processor API and
remove the
operator which belongs to this statement or you not using the savepoint at
all (no savepoint restore).
Be aware that these ways you're going to loose state data so be careful.

If you have the possibility to invest more time then with the state
processor API you
can do the state migration where you don't loose anything.

BR,
G


On Sat, Feb 8, 2025 at 7:40 AM 李 琳 <le...@outlook.com> wrote:

> Dear All,
>
> In a Flink job using statementSet, I have two DMLs for index calculation.
> I made a modification to one of the DMLs, changing the group keys from "*distinct_id,
> $city, $province, window_start, window_end*" to "*distinct_id,
> window_start, window_end*". When attempting to restore the job from a
> savepoint, Flink throws an exception due to an incompatibility in the key
> serializer between the original and the new state.
>
> I'm seeking a way to make Flink skip the state related to the incompatible
> key serializer and proceed with restoring the unmodified DMLs. Are there
> any more effective approaches to achieve this?
>
> Best regards, leilinee
>
>