LeftOuterJoin in Flink (JAVA API)

Question

我试图在Flink中做一个LeftOuterJoin。我没有尝试自己实现leftOuterJoin此处使用CoGroupFunction: https://gist.github.com/mxm/c2e9c459a9d82c18d789

我正在尝试使用FlatJoinFunction:

    public static final class leftOuter implements FlatJoinFunction<Tuple3<String,String,String>, Tuple2<String,String>, Tuple2<String,String>>{

    @Override
    public void join(Tuple3<String, String, String> in1,
            Tuple2<String, String> in2,
            Collector<Tuple2<String, String>> out) throws Exception {
        // TODO Auto-generated method stub
        out.collect(new Tuple2<String,String>(in1.f0, in2.f1 == null ? "null" : in2.f1));
    }
}

我这样调用这个函数:

        input1.leftOuterJoin(input2).where(0)
            .equalTo(1)
            .with(new leftOuter());

不幸的是，我得到一个NullPointerException在输出。收集行。

提前感谢您的帮助!

Answer 1

这是左外连接的预期行为。

给定程序，左外连接在两种情况下调用JoinFunction:

1. 如果两个输入，input1和input2有相同连接键的记录，则对该键的笛卡尔积的每个元素调用join()。
2. 如果左输入input1有右输入(input2)中没有的键记录，则join()会被input1和null的每个键记录调用。

您应该在JoinFunction中添加in2 == null的检查