Flink TypeInfo Java Generics

Question

我希望我的Flink应用程序使用Flinkorg.apache.flink.api.java.tuple.Tuple2类将Kafka主题中的数据反序列化为ConsumerRecord<byte[], byte[]>的Flink流或Tuple2<byte[], byte[]>的Flint流。

使用这两个选项，我无法获得编译的getProducedType实现。

public class KafkaNOOPDeserialization implements KafkaDeserializationSchema<ConsumerRecord<byte[], byte[]>> {
@Override
public boolean isEndOfStream(ConsumerRecord<byte[], byte[]> nextElement) {
return false;
}
@Override
public ConsumerRecord<byte[], byte[]> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
return record;
}
@Override
public TypeInformation<ConsumerRecord<byte[], byte[]>> getProducedType() {
// This provides TypeInformation<ConsumerRecord>
// It needs TypeInformation<ConsumerRecord<byte[], byte[]>>
var typeInfo = TypeExtractor.getForClass(ConsumerRecord.class);
// This hard cast won't compile
return (TypeInformation<ConsumerRecord<byte[], byte[]>>) typeInfo;
}
}

public class KafkaByteArrayTupleDeserializer implements KafkaDeserializationSchema<Tuple2<byte[], byte[]>> {
@Override
public boolean isEndOfStream(Tuple2<byte[], byte[]> nextElement) {
return false;
}
@Override
public Tuple2<byte[], byte[]> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
return new Tuple2(record.key(), record.value());
}
@Override
public TypeInformation<Tuple2<byte[], byte[]>> getProducedType() {
// This provides TupleTypeInfo<Tuple>
// It needs TupleTypeInfo<Tuple2<byte[], byte[]>>
var typeInfo = TupleTypeInfo.getBasicAndBasicValueTupleTypeInfo(
byte[].class, byte[].class);
// Hard cast won't compile
return (TypeInformation<Tuple2<byte[], byte[]>>) typeInfo;
}
}

Answer 1

Flink提供了org.apache.flink.api.common.typeinfo.TypeHint类来帮助实现这一点。(文档(

例如，在您的第二个示例中，而不是：

return (TypeInformation<Tuple2<byte[], byte[]>>) typeInfo;

你可以写：

return new TypeHint<Tuple2<byte[], byte[]>>(){}.getTypeInfo();