不懂间隔加入Flink

Question

来自Flink的官方文件：

https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/joining.html#interval-join

示例代码是：

import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;

...

val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...

orangeStream
    .keyBy(elem => /* select key */)
    .intervalJoin(greenStream.keyBy(elem => /* select key */))
    .between(Time.milliseconds(-2), Time.milliseconds(1))
    .process(new ProcessJoinFunction[Integer, Integer, String] {
        override def processElement(left: Integer, right: Integer, ctx: ProcessJoinFunction[Integer, Integer, String]#Context, out: Collector[String]): Unit = {
         out.collect(left + "," + right); 
        }
      });
    });

从上面的代码中，我想知道如何指定执行此间隔连接的开始时间（例如，从今天开始）（开始时间之前的数据将不考虑在内）。

例如，我已经运行了3天的程序，我不想对所有数据执行3天的连接，我只想对今天生成的数据执行连接。

Answer 1

我不认为它像你想象的那样工作。

在本例中，实际间隔是根据＜code＞orangeStream＜/code＞的实际时间戳计算的，因此您并没有真正提供要考虑的数据间隔，而是类似于窗口，指定哪些元素将与OrangeStriam的给定元素连接。

因此，对于上述窗口，如果您有时间戳为5的橙色元素，则它将与时间戳为3到6的元素连接。

我真的不认为您可以使用它仅对部分数据执行连接，我唯一能想到的是使用时间戳过滤数据，并过滤掉之前生成的所有元素。