Hey, IMHO, the simplest way in your case would be to use the Evictor to evict duplicate values after the window is generated. Have look at it here: https://ci.apache.org/projects/flink/flink-docs-release-1.6/api/java/org/apache/flink/streaming/api/windowing/evictors/Evictor.html
Best Regards, Dominik. pon., 8 paź 2018 o 08:00 Eric L Goodman <eric.good...@colorado.edu> napisał(a): > What is the best way to avoid or remove duplicates when joining a stream > with itself? I'm performing a streaming temporal triangle computation and > the first part is to find triads of two edges of the form vertexA->vertexB > and vertexB->vertexC (and there are temporal constraints where the first > edge occurs before the second edge). To do that, I have the following code: > > DataStream<Triad> triads = edges.join(edges) > .where(new DestKeySelector()) > .equalTo(new SourceKeySelector()) > .window(SlidingEventTimeWindows.of(Time.milliseconds(windowSizeMs), > Time.milliseconds(slideSizeMs))) > .apply(new EdgeJoiner(queryWindow)); > > However, when I look at the triads being built, there are two copies of each > triad. > > For example, if I create ten edges (time, source, target): > > 0.0, 4, 0 > > 0.01, 1, 5 > > 0.02, 3, 7 > > 0.03, 0, 8 > > 0.04, 0, 9 > > 0.05, 4, 8 > > 0.06, 4, 3 > > 0.07, 5, 9 > > 0.08, 7, 1 > > 0.09, 9, 6 > > > It creates the following triads (time1, source1, target1, time2, source2, > targe2). Note there are two copies of each. > > 0.0, 4, 0 0.03, 0, 8 > > 0.0, 4, 0 0.03, 0, 8 > > 0.0, 4, 0 0.04, 0, 9 > > 0.0, 4, 0 0.04, 0, 9 > > 0.01, 1, 5 0.07, 5, 9 > > 0.01, 1, 5 0.07, 5, 9 > > 0.02, 3, 7 0.08, 7, 1 > > 0.02, 3, 7 0.08, 7, 1 > > 0.04, 0, 9 0.09, 9, 6 > > 0.04, 0, 9 0.09, 9, 6 > > 0.07, 5, 9 0.09, 9, 6 > > 0.07, 5, 9 0.09, 9, 6 > > I'm assuming this behavior has something to do with the joining of "edges" > with itself. > > I can provide more code if that would be helpful, but I believe I've captured > the most salient portion. > > > > > >
