[
https://issues.apache.org/jira/browse/FLINK-16596?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17059309#comment-17059309
]
Felix Wollschläger edited comment on FLINK-16596 at 3/14/20, 11:43 AM:
-----------------------------------------------------------------------
A simple solution for this would be:
Add a default-interface-Method to the KeySelector interface so users can still
use, for example, Tuple-Keys with an Enum, but let the user provide a custom
hashCode-Implementation:
{code}
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.api.java.functions;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.Function;
import java.io.Serializable;
/**
* The {@link KeySelector} allows to use deterministic objects for operations
such as
* reduce, reduceGroup, join, coGroup, etc. If invoked multiple times on the
same object,
* the returned key must be the same.
*
* <p>The extractor takes an object and returns the deterministic key for that
object.
*
* @param <IN> Type of objects to extract the key from.
* @param <KEY> Type of key.
*/
@Public
@FunctionalInterface
public interface KeySelector<IN, KEY> extends Function, Serializable {
/**
* User-defined function that deterministically extracts the key from
an object.
*
* <p>For example for a class:
* <pre>
* public class Word {
* String word;
* int count;
* }
* </pre>
* The key extractor could return the word as
* a key to group all Word objects by the String they contain.
*
* <p>The code would look like this
* <pre>
* public String getKey(Word w) {
* return w.word;
* }
* </pre>
*
* @param value The object to get the key from.
* @return The extracted key.
*
* @throws Exception Throwing an exception will cause the execution of
the respective task to fail,
* and trigger recovery or cancellation of the
program.
*/
KEY getKey(IN value) throws Exception;
/**
* Attention: When implementing this method be sure to not break the
contract of equals/hashCode.
* <pre>
* // Example
* // if
* keyA.equals(keyB)
* // ensure that
* this.buildHashCodeForKey(keyA) == this.buildHashCodeForKey(keyB)
* </pre>
* @param key The key to build a hashCode for (the default
implementation returns the hashCode of the supplied key)
* @return The hashCode of the key
* @throws Exception Throwing an exception will cause the execution of
the respective task to fail,
* and trigger recovery or cancellation of the
program.
*/
default int buildHashCodeForKey(KEY key) throws Exception {
return key.hashCode();
}
}
{code}
And change the implementation of HashSelector to use the new method:
{code}
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.state.api.output.partitioner;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.util.Preconditions;
/**
* A wrapper around a {@link KeySelector} that returns the {@link
Object#hashCode()} of the returned
* key.
*
* @param <IN> Type of objects to extract the key from.
* @param <KEY> Type of the extracted key
*/
@Internal
public class HashSelector<IN, KEY> implements KeySelector<IN, Integer> {
private static final long serialVersionUID = 1L;
private final KeySelector<IN, KEY> keySelector;
public HashSelector(KeySelector<IN, KEY> keySelector) {
Preconditions.checkNotNull(keySelector);
this.keySelector = keySelector;
}
@Override
public Integer getKey(IN value) throws Exception {
return
keySelector.buildHashCodeForKey(keySelector.getKey(value));
}
}
{code}
was (Author: fwollsch):
A simple solution for this would be:
Add a default-interface-Method to the KeySelector interface so users can still
use, for example, Tuple-Keys with an Enum, but let the user provide a custom
hashCode-Implementation:
{code}
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.api.java.functions;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.Function;
import java.io.Serializable;
/**
* The {@link KeySelector} allows to use deterministic objects for operations
such as
* reduce, reduceGroup, join, coGroup, etc. If invoked multiple times on the
same object,
* the returned key must be the same.
*
* <p>The extractor takes an object and returns the deterministic key for that
object.
*
* @param <IN> Type of objects to extract the key from.
* @param <KEY> Type of key.
*/
@Public
@FunctionalInterface
public interface KeySelector<IN, KEY> extends Function, Serializable {
/**
* User-defined function that deterministically extracts the key from
an object.
*
* <p>For example for a class:
* <pre>
* public class Word {
* String word;
* int count;
* }
* </pre>
* The key extractor could return the word as
* a key to group all Word objects by the String they contain.
*
* <p>The code would look like this
* <pre>
* public String getKey(Word w) {
* return w.word;
* }
* </pre>
*
* @param value The object to get the key from.
* @return The extracted key.
*
* @throws Exception Throwing an exception will cause the execution of
the respective task to fail,
* and trigger recovery or cancellation of the
program.
*/
KEY getKey(IN value) throws Exception;
default int buildHashCodeForKey(KEY key) throws Exception {
return key.hashCode();
}
}
{code}
And change the implementation of HashSelector to use the new method:
{code}
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.flink.state.api.output.partitioner;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.util.Preconditions;
/**
* A wrapper around a {@link KeySelector} that returns the {@link
Object#hashCode()} of the returned
* key.
*
* @param <IN> Type of objects to extract the key from.
* @param <KEY> Type of the extracted key
*/
@Internal
public class HashSelector<IN, KEY> implements KeySelector<IN, Integer> {
private static final long serialVersionUID = 1L;
private final KeySelector<IN, KEY> keySelector;
public HashSelector(KeySelector<IN, KEY> keySelector) {
Preconditions.checkNotNull(keySelector);
this.keySelector = keySelector;
}
@Override
public Integer getKey(IN value) throws Exception {
return
keySelector.buildHashCodeForKey(keySelector.getKey(value));
}
}
{code}
> Support Enum-Values as part of a Key
> ------------------------------------
>
> Key: FLINK-16596
> URL: https://issues.apache.org/jira/browse/FLINK-16596
> Project: Flink
> Issue Type: Improvement
> Reporter: Felix Wollschläger
> Priority: Major
>
> See: FLINK-11774
> h2. Description:
> The hashCode implementation of Enum-Values is guaranteed to be stable inside
> a JVM, but not accross multiple JVMs. This leads to failures when restoring
> from a checkpoint/savepoint containing Keyed-State on Keys with Enums as a
> part of the key.
> For users of Flink there is a workaround to solve this problem:
> Don't rely on the hashCode-Implementation of Enum-Values in the
> hashCode-Implementation of the actual key. Use the ordinal() or
> name().hashCode() instead of the Enum hashCode.
> h2. Goals of this improvement:
> Implement a way to handle Enum-Values, by either handling Enum-Values on a
> internal level or implementing a abstract Base KeySelector class that users
> of flink can choose if their key contains a Enum-Value.
> Code to reproduce a failure and first thoughts can be found in FLINK-11774 .
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
