@Test public void testComm() throws Exception { PCollection<Tuple3<String, String, String>> comm = Set.comm(set1, set2); Iterator<Tuple3<String, String, String>> i = comm.materialize().iterator(); checkEquals(null, null, "a", i.next()); checkEquals("b", null, null, i.next()); checkEquals(null, null, "c", i.next()); checkEquals(null, "d", null, i.next()); checkEquals("e", null, null, i.next()); assertFalse(i.hasNext()); }
public static <T> PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1, PCollection<T> coll2) { PTypeFamily typeFamily = coll1.getTypeFamily(); PType<T> type = coll1.getPType(); return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, Tuple3<T, T, T>>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<Tuple3<T, T, T>> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); boolean inFirst = !groups.first().isEmpty(); boolean inSecond = !groups.second().isEmpty(); T t = input.first(); emitter.emit(Tuple3.of( inFirst && !inSecond ? t : null, !inFirst && inSecond ? t : null, inFirst && inSecond ? t : null)); } }, typeFamily.triples(type, type, type)); }
Set { public static <T> PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1, PCollection<T> coll2) { PTypeFamily typeFamily = coll1.getTypeFamily(); PType<T> type = coll1.getPType(); return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, Tuple3<T, T, T>>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<Tuple3<T, T, T>> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); boolean inFirst = !groups.first().isEmpty(); boolean inSecond = !groups.second().isEmpty(); T t = input.first(); emitter.emit(Tuple3.of( inFirst && !inSecond ? t : null, !inFirst && inSecond ? t : null, inFirst && inSecond ? t : null)); } }, typeFamily.triples(type, type, type)); } }
Set { public static <T> PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1, PCollection<T> coll2) { PTypeFamily typeFamily = coll1.getTypeFamily(); PType<T> type = coll1.getPType(); return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, Tuple3<T, T, T>>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<Tuple3<T, T, T>> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); boolean inFirst = !groups.first().isEmpty(); boolean inSecond = !groups.second().isEmpty(); T t = input.first(); emitter.emit(Tuple3.of( inFirst && !inSecond ? t : null, !inFirst && inSecond ? t : null, inFirst && inSecond ? t : null)); } }, typeFamily.triples(type, type, type)); }  }
Set { public static <T> PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1, PCollection<T> coll2) { PTypeFamily typeFamily = coll1.getTypeFamily(); PType<T> type = coll1.getPType(); return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, Tuple3<T, T, T>>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<Tuple3<T, T, T>> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); boolean inFirst = !groups.first().isEmpty(); boolean inSecond = !groups.second().isEmpty(); T t = input.first(); emitter.emit(Tuple3.of( inFirst && !inSecond ? t : null, !inFirst && inSecond ? t : null, inFirst && inSecond ? t : null)); } }, typeFamily.triples(type, type, type)); }  static PCollection<T> difference(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<T> intersection(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1,
      PCollection<T> coll2); }
Set { public static <T> PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1, PCollection<T> coll2) { PTypeFamily typeFamily = coll1.getTypeFamily(); PType<T> type = coll1.getPType(); return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, Tuple3<T, T, T>>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<Tuple3<T, T, T>> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); boolean inFirst = !groups.first().isEmpty(); boolean inSecond = !groups.second().isEmpty(); T t = input.first(); emitter.emit(Tuple3.of( inFirst && !inSecond ? t : null, !inFirst && inSecond ? t : null, inFirst && inSecond ? t : null)); } }, typeFamily.triples(type, type, type)); }  static PCollection<T> difference(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<T> intersection(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1,
      PCollection<T> coll2);  }