Flink：Flink是否支持可以处理具有公共字段的不同数据流的抽象运算符？

Question

假设我们有多个数据流，它们共享一些共同的特性。

例如，我们有一个教师流和一个学生流，它们都有一个年龄字段。如果我想从实时流中找出最大的学生或老师，我可以实现一个运算符，如下所示。

public MaxiumAgeFunc extends RichMapFunction<Student,Integer> {
    int maxAge;

    @Override
    public void flatMap(Student s, Collector<Integer> collector) throws Exception {
        if(s.age > maxAge){
            maxAge = s.age;
        }
        collector.collect(maxAge);
    }
}

为了找出最年长的老师，我们需要实现一个类似的操作符，如下所示

public MaxiumAgeFunc extends RichMapFunction<Teacher,Integer> {
    int maxAge;

    @Override
    public void flatMap(Teacher t, Collector<Integer> collector) throws Exception {
        if(t.age > maxAge){
            maxAge = t.age;
        }
        collector.collect(maxAge);
    }
}

但实际上这两个操作符有共同的流程逻辑，所以我的想法是定义一个父类，比如People。

public class People{
    public Integer age;
}

然后学生和教师可以定义为他们的子类，也可以保留自己的字段。

public class Student extends People {
    public Integer grade;  // student grade
    ...
}

public class Student extends People {
    public Integer subject;  // the subject that teacher teaches
    ...
}

在这种情况下，我可以定义一个操作符，如下所示。

public MaxiumAgeFunc extends RichMapFunction<People,Integer> {
    int maxAge;

    @Override
    public void flatMap(People p, Collector<Integer> collector) throws Exception {
        if(t.age > maxAge){
            maxAge = p.age;
        }
        collector.collect(maxAge);
    }
}

但是，当我尝试使用此操作符实现Flink执行拓扑时，由于数据类型不匹配，它将无法工作。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Student> studentStream = env.addSource(...);
DataStream<Teacher> teacherStream = env.addSource(...);

studentStream.map(new MaxiumAgeFunc()).print();
teacherStream.map(new MaxiumAgeFunc()).print();

这就是我的问题，有没有可能为具有公共字段的输入流创建一个抽象运算符？

Answer 1

这比Flink问题更Java：

您要做的是将MaxiumAgeFunc参数化如下

public MaxiumAgeFunc<T extends People> extends RichMapFunction<T, Integer> {
    int maxAge;

    @Override
    public void flatMap(T p, Collector<Integer> collector) throws Exception {
        if(t.age > maxAge){
            maxAge = p.age;
        }
        collector.collect(maxAge);
    }
}

然后像这样使用它

studentStream.map(new MaxiumAgeFunc<>()).print();
teacherStream.map(new MaxiumAgeFunc<>()).print();

编辑：

顺便说一句，您的函数没有使用检查点（因此从检查点恢复时会产生错误的结果），我宁愿使用全局窗口上的聚合函数。

students
    .windowAll(GlobalWindows.create())
    .aggregate(new AggregateFunction<People, Integer, Integer>() {
        @Override
        public Integer createAccumulator() {
            return -1;
        }

        @Override
        public Integer add(People value, Integer accumulator) {
            return Math.max(value.age, accumulator);
        }

        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return Math.max(a, b);
        }
    });