Apache Flink - 使用具有泛型类型参数的类答案

【问题标题】：Apache Flink - Using class with generic type parameterApache Flink - 使用具有泛型类型参数的类
【发布时间】：2018-05-23 05:52:55
【问题描述】：

如何在 flink 中使用具有泛型类型的类？我遇到了错误：

由于类型擦除，无法自动确定函数“main(StreamingJob.java:63)”的返回类型。您可以通过对转换调用的结果使用returns(...) 方法或让您的函数实现“ResultTypeQueryable”接口来提供类型信息提示。

我使用的类是这样的：

class MaybeProcessable<T> {
    private final T value;

    public MaybeProcessable(T value) {
        this.value = value;
    }

    public T get() {
        return value;
    }
}

我正在使用一个示例 flink 作业，例如：

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

    env.addSource(new PubSubSource(PROJECT_ID, SUBSCRIPTION_NAME))
        .map(MaybeProcessable::new)
        .map(MaybeProcessable::get)
        .writeAsText("/tmp/flink-output", FileSystem.WriteMode.OVERWRITE);

    // execute program
    env.execute("Flink Streaming Java API Skeleton");
}

现在我可以使用 .returns() 函数添加一个 TypeInformation 实例：

.map(MaybeProcessable::new).returns(new MyCustomTypeInformationClass(String.class))

但这需要我编写自己的序列化程序。有没有更简单的方法来实现这一点？

【问题讨论】：

你能不能创建一个非泛型子类：class StringX extends MaybeProcessable<String>？
很好，对于某些功能，我可以做到这一点。但在其他情况下，我想重用 Map 函数，因此不让它们在 StringX 上工作

标签： java generics java-8 apache-flink

【解决方案1】：

你可以使用 .returns(TypeInformation.of(new TypeHint<MaybeProcessing<#CONCRETE_TYPE_HERE>>{}) 每次重用泛型类型的 MapFunction 来设置返回类型，而无需创建您自己的任何自定义类。

【讨论】：