java - Flink 中的 LeftOuterJoin (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());

不幸的是，我在 out.collect 行中遇到了 NullPointerException。

预先感谢您的帮助!

Answer 1

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

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

1. 如果两个输入 input1 和 input2 都有具有相同连接键的记录，则为每个元素调用 join()此 key 的笛卡尔积。
2. 如果左侧输入 input1 的记录的键在右侧输入 (input2) 中不存在，则 join() 为每条记录调用 input1 键，为正确的输入调用 null。

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