Flink 0.10.0 最近刚刚发布。我需要从 0.9.1 迁移一些代码。但出现以下错误：
org.apache.flink.api.common.functions.InvalidTypesException：无法确定“class fi.aalto.dmg.frame.FlinkPairWorkloadOperator”中类型变量“K”的类型。这很可能是类型擦除问题。类型提取目前仅在返回类型中的所有变量都可以从输入类型推导出来的情况下支持具有泛型变量的类型。
这是代码：
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);
}
};
}