@Test public void testDifference() throws Exception { PCollection<String> difference = Set.difference(set1, set2); assertEquals(Lists.newArrayList("b", "e"), Lists.newArrayList(difference.materialize())); }
public static <T> PCollection<T> difference(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); }
Set { public static <T> PCollection<T> difference(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); } }
Set { public static <T> PCollection<T> difference(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); }  }
Set { public static <T> PCollection<T> difference(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); }  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<T> difference(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); }  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);  }