Yes it is possible to build an adjacency list with a reduce. But I don't see how I can replace the coGroup with a join. The new JoinFunction interface does not provide a collector as output but can only emit one tuple. I think we need an additional "FlatJoinFunction" or something with an collector. I opened an issue to discuss this: #662