Flink InvalidTypesException：无法确定 'class' 中的 TypeVariable 'K' 的类型

Question

Flink 0.10.0 刚刚发布。我有一些代码需要从 0.9.1 迁移。但是出现以下错误：

org.apache.flink.api.common.functions.InvalidTypesException：无法确定 'class fi.aalto.dmg.frame.FlinkPairWorkloadOperator' 中 TypeVariable 'K' 的类型。这很可能是类型擦除问题。类型提取当前仅在 return 类型中的所有变量都可以从输入类型中推导出来的情况下才支持具有通用变量的类型。

代码如下：

 public class FlinkPairWorkloadOperator<K,V> implements PairWorkloadOperator<K,V> {

    private DataStream<Tuple2<K, V>> dataStream;

    public FlinkPairWorkloadOperator(DataStream<Tuple2<K, V>> dataStream1) {
        this.dataStream = dataStream1;
    }



    public FlinkGroupedWorkloadOperator<K, V> groupByKey() {
        KeyedStream<Tuple2<K, V>, K> keyedStream = this.dataStream.keyBy(new KeySelector<Tuple2<K, V>, K>() {
            @Override
            public K getKey(Tuple2<K, V> value) throws Exception {
                return value._1();
            }
        });
        return new FlinkGroupedWorkloadOperator<>(keyedStream);
    }
}

为了理解 InvalidTypesException 是如何发生的，我有另一个例子也抛出了这个异常，但我对此一无所知。在此演示中，程序使用 scala.Tuple2，但不使用 flink Tuple2。

public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> counts = env
            .socketTextStream("localhost", 9999)
            .flatMap(new Splitter());

        DataStream<Tuple2<String, Integer>> pairs = mapToPair(counts, mapToStringIntegerPair);
        pairs.print();
        env.execute("Socket Stream WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String sentence, Collector<String> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(word);
            }
        }
    }

    public static  <K,V,T> DataStream<Tuple2<K,V>> mapToPair(DataStream<T> dataStream , final MapPairFunction<T, K, V> fun){
        return dataStream.map(new MapFunction<T, Tuple2<K, V>>() {
            @Override
            public Tuple2<K, V> map(T t) throws Exception {
                return fun.mapPair(t);
            }
        });
    }

   public interface MapPairFunction<T, K, V> extends Serializable {
     Tuple2<K,V> mapPair(T t);
  }

  public static MapPairFunction<String, String, Integer> mapToStringIntegerPair = new MapPairFunction<String, String, Integer>() {
       public Tuple2<String, Integer> mapPair(String s) {
            return new Tuple2<String, Integer>(s, 1);
        }
    };
}

Answer 1

问题是您使用 scala.Tuple2 而不是 org.apache.flink.api.java.tuple.Tuple2 与 Flink 的 Java API 结合使用。 Java API 的 TypeExtractor 不理解 Scala 元组。因此，它无法提取类型变量 K.

的类型

如果您改用 org.apache.flink.api.java.tuple.Tuple2，则 TypeExtractor 将能够解析类型变量。

Flink InvalidTypesException：无法确定 'class' 中的 TypeVariable 'K' 的类型

Flink InvalidTypesException: Type of TypeVariable 'K' in 'class' could not be determined

apache-flink

flink-streaming