git.net

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Duplicates in self join


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.