@Test public void testCoGroup() throws Exception { VertexFactory<EPGMVertex> vertexFactory = getConfig().getLogicalGraphFactory() .getVertexFactory(); EPGMVertex v1 = vertexFactory.createVertex("a"); Tuple2<GradoopId, PropertyValue> property1 = Tuple2.of(v1.getId(), PropertyValue.create(1L)); EPGMVertex v2 = vertexFactory.createVertex("a"); EPGMVertex v3 = vertexFactory.createVertex("b"); Tuple2<GradoopId, PropertyValue> property2 = Tuple2.of(v3.getId(), PropertyValue.create(1L)); List<EPGMVertex> input = Arrays.asList(v1, v2, v3); List<EPGMVertex> result = getExecutionEnvironment().fromElements(property1, property2) .coGroup(getExecutionEnvironment().fromCollection(input)) .where(0).equalTo(new Id<>()) .with(new AccumulatePropagatedValues<>("k", Collections.singleton("a"))) .collect(); v1.setProperty("k", PropertyValue.create(Collections.singletonList(PropertyValue.create(1L)))); List<EPGMVertex> expected = Arrays.asList(v1, v2, v3); Comparator<EPGMVertex> comparator = Comparator.comparing(EPGMVertex::getId); expected.sort(comparator); result.sort(comparator); assertArrayEquals(expected.toArray(), result.toArray()); }
@Override public void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues, Iterable<V> elements, Collector<V> out) { Iterator<V> iterator = elements.iterator(); if (!iterator.hasNext()) { return; } V targetVertex = iterator.next(); if (targetVertexLabels != null && !targetVertexLabels.contains(targetVertex.getLabel())) { out.collect(targetVertex); return; } List<PropertyValue> values = new ArrayList<>(); propertyValues.forEach(t -> values.add(t.f1)); if (!values.isEmpty()) { targetVertex.setProperty(targetVertexPropertyKey, PropertyValue.create(values)); } out.collect(targetVertex); }
AccumulatePropagatedValues implements CoGroupFunction<Tuple2<GradoopId, PropertyValue>, V, V> { @Override public void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues, Iterable<V> elements, Collector<V> out) { Iterator<V> iterator = elements.iterator(); if (!iterator.hasNext()) { return; } V targetVertex = iterator.next(); if (targetVertexLabels != null && !targetVertexLabels.contains(targetVertex.getLabel())) { out.collect(targetVertex); return; } List<PropertyValue> values = new ArrayList<>(); propertyValues.forEach(t -> values.add(t.f1)); if (!values.isEmpty()) { targetVertex.setProperty(targetVertexPropertyKey, PropertyValue.create(values)); } out.collect(targetVertex); } }
AccumulatePropagatedValues implements CoGroupFunction<Tuple2<GradoopId, PropertyValue>, V, V> { @Override public void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues, Iterable<V> elements, Collector<V> out) { Iterator<V> iterator = elements.iterator(); if (!iterator.hasNext()) { return; } V targetVertex = iterator.next(); if (targetVertexLabels != null && !targetVertexLabels.contains(targetVertex.getLabel())) { out.collect(targetVertex); return; } List<PropertyValue> values = new ArrayList<>(); propertyValues.forEach(t -> values.add(t.f1)); if (!values.isEmpty()) { targetVertex.setProperty(targetVertexPropertyKey, PropertyValue.create(values)); } out.collect(targetVertex); } AccumulatePropagatedValues(String targetVertexPropertyKey,
    Set<String> targetVertexLabels); }
AccumulatePropagatedValues implements CoGroupFunction<Tuple2<GradoopId, PropertyValue>, V, V> { @Override public void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues, Iterable<V> elements, Collector<V> out) { Iterator<V> iterator = elements.iterator(); if (!iterator.hasNext()) { return; } V targetVertex = iterator.next(); if (targetVertexLabels != null && !targetVertexLabels.contains(targetVertex.getLabel())) { out.collect(targetVertex); return; } List<PropertyValue> values = new ArrayList<>(); propertyValues.forEach(t -> values.add(t.f1)); if (!values.isEmpty()) { targetVertex.setProperty(targetVertexPropertyKey, PropertyValue.create(values)); } out.collect(targetVertex); } AccumulatePropagatedValues(String targetVertexPropertyKey,
    Set<String> targetVertexLabels); @Override void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues,
                      Iterable<V> elements, Collector<V> out); }
AccumulatePropagatedValues implements CoGroupFunction<Tuple2<GradoopId, PropertyValue>, V, V> { @Override public void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues, Iterable<V> elements, Collector<V> out) { Iterator<V> iterator = elements.iterator(); if (!iterator.hasNext()) { return; } V targetVertex = iterator.next(); if (targetVertexLabels != null && !targetVertexLabels.contains(targetVertex.getLabel())) { out.collect(targetVertex); return; } List<PropertyValue> values = new ArrayList<>(); propertyValues.forEach(t -> values.add(t.f1)); if (!values.isEmpty()) { targetVertex.setProperty(targetVertexPropertyKey, PropertyValue.create(values)); } out.collect(targetVertex); } AccumulatePropagatedValues(String targetVertexPropertyKey,
    Set<String> targetVertexLabels); @Override void coGroup(Iterable<Tuple2<GradoopId, PropertyValue>> propertyValues,
                      Iterable<V> elements, Collector<V> out);  }