@Test public void testWithNonEmptyNeighborhood() throws Exception { EPGMVertex someVertex = vertexFactory.createVertex(); final int count = 10; List<GradoopId> ids = Stream.generate(GradoopId::get).limit(count).collect(Collectors.toList()); List<NeighborhoodVertex> vertexPojos = ids.stream() .map(id -> new NeighborhoodVertex(id, "")) .collect(Collectors.toList()); Tuple2<EPGMVertex, List<NeighborhoodVertex>> inputNonEmpty = new Tuple2<>(someVertex, vertexPojos); List<EPGMEdge> result = getExecutionEnvironment().fromElements(inputNonEmpty) .flatMap(toTest).collect(); assertEquals(count * (count - 1), result.size()); for (EPGMEdge edge : result) { assertNotEquals(edge.getSourceId(), edge.getTargetId()); } long disctinctCount = result.stream() .map(e -> new Tuple2<>(e.getSourceId(), e.getTargetId())) .distinct().count(); assertEquals((long) result.size(), disctinctCount); }
@Override public void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out) { final List<NeighborhoodVertex> neighbors = value.f1; for (NeighborhoodVertex source : neighbors) { reuseEdge.setSourceId(source.getNeighborId()); for (NeighborhoodVertex target : neighbors) { if (source == target) { continue; } reuseEdge.setId(GradoopId.get()); reuseEdge.setTargetId(target.getNeighborId()); out.collect(reuseEdge); } } }
CreateCartesianNeighborhoodEdges implements FlatMapFunction<Tuple2<V, List<NeighborhoodVertex>>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out) { final List<NeighborhoodVertex> neighbors = value.f1; for (NeighborhoodVertex source : neighbors) { reuseEdge.setSourceId(source.getNeighborId()); for (NeighborhoodVertex target : neighbors) { if (source == target) { continue; } reuseEdge.setId(GradoopId.get()); reuseEdge.setTargetId(target.getNeighborId()); out.collect(reuseEdge); } } } }
CreateCartesianNeighborhoodEdges implements FlatMapFunction<Tuple2<V, List<NeighborhoodVertex>>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out) { final List<NeighborhoodVertex> neighbors = value.f1; for (NeighborhoodVertex source : neighbors) { reuseEdge.setSourceId(source.getNeighborId()); for (NeighborhoodVertex target : neighbors) { if (source == target) { continue; } reuseEdge.setId(GradoopId.get()); reuseEdge.setTargetId(target.getNeighborId()); out.collect(reuseEdge); } } } CreateCartesianNeighborhoodEdges(EdgeFactory<E> factory, String newEdgeLabel); }
CreateCartesianNeighborhoodEdges implements FlatMapFunction<Tuple2<V, List<NeighborhoodVertex>>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out) { final List<NeighborhoodVertex> neighbors = value.f1; for (NeighborhoodVertex source : neighbors) { reuseEdge.setSourceId(source.getNeighborId()); for (NeighborhoodVertex target : neighbors) { if (source == target) { continue; } reuseEdge.setId(GradoopId.get()); reuseEdge.setTargetId(target.getNeighborId()); out.collect(reuseEdge); } } } CreateCartesianNeighborhoodEdges(EdgeFactory<E> factory, String newEdgeLabel); @Override void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out); @Override TypeInformation<E> getProducedType(); }
CreateCartesianNeighborhoodEdges implements FlatMapFunction<Tuple2<V, List<NeighborhoodVertex>>, E>, ResultTypeQueryable<E> { @Override public void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out) { final List<NeighborhoodVertex> neighbors = value.f1; for (NeighborhoodVertex source : neighbors) { reuseEdge.setSourceId(source.getNeighborId()); for (NeighborhoodVertex target : neighbors) { if (source == target) { continue; } reuseEdge.setId(GradoopId.get()); reuseEdge.setTargetId(target.getNeighborId()); out.collect(reuseEdge); } } } CreateCartesianNeighborhoodEdges(EdgeFactory<E> factory, String newEdgeLabel); @Override void flatMap(Tuple2<V, List<NeighborhoodVertex>> value, Collector<E> out); @Override TypeInformation<E> getProducedType();  }