Filter extends PTransform<PCollection<T>, PCollection<T>> { public static <T, PredicateT extends ProcessFunction<T, Boolean>> Filter<T> by( PredicateT predicate) { return new Filter<>(predicate); } private  Filter(ProcessFunction<T, Boolean> predicate); private  Filter(ProcessFunction<T, Boolean> predicate, String predicateDescription); static Filter<T> by(
      PredicateT predicate); static Filter<T> by(
      PredicateT predicate); static Filter<T> lessThan(final T value); static Filter<T> greaterThan(final T value); static Filter<T> lessThanEq(final T value); static Filter<T> greaterThanEq(final T value); static Filter<T> equal(final T value); @Override PCollection<T> expand(PCollection<T> input); @Override void populateDisplayData(DisplayData.Builder builder);  }
@Test @Category(NeedsRunner.class) public void testIdentityFilterByPredicate() { PCollection<Integer> output = p.apply(Create.of(591, 11789, 1257, 24578, 24799, 307)) .apply(Filter.by(new TrivialFn(true))); PAssert.that(output).containsInAnyOrder(591, 11789, 1257, 24578, 24799, 307); p.run(); }
@Test @Category(NeedsRunner.class) public void testNoFilterByPredicate() { PCollection<Integer> output = p.apply(Create.of(1, 2, 4, 5)).apply(Filter.by(new TrivialFn(false))); PAssert.that(output).empty(); p.run(); }
@Test @Category(NeedsRunner.class) public void testFilterByPredicate() { PCollection<Integer> output = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7)).apply(Filter.by(new EvenFn())); PAssert.that(output).containsInAnyOrder(2, 4, 6); p.run(); }
@Test @Category(NeedsRunner.class) public void testFilterByProcessFunction() { PCollection<Integer> output = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7)).apply(Filter.by(new EvenProcessFn())); PAssert.that(output).containsInAnyOrder(2, 4, 6); p.run(); }
@Test @Category(NeedsRunner.class) public void testIdentityFilterByPredicateWithLambda() { PCollection<Integer> output = p.apply(Create.of(591, 11789, 1257, 24578, 24799, 307)).apply(Filter.by(i -> true)); PAssert.that(output).containsInAnyOrder(591, 11789, 1257, 24578, 24799, 307); p.run(); }
@Test @Category(NeedsRunner.class) public void testNoFilterByPredicateWithLambda() { PCollection<Integer> output = p.apply(Create.of(1, 2, 4, 5)).apply(Filter.by(i -> false)); PAssert.that(output).empty(); p.run(); }
@Test @Category(NeedsRunner.class) public void testFilterByPredicateWithLambda() { PCollection<Integer> output = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7)).apply(Filter.by(i -> i % 2 == 0)); PAssert.that(output).containsInAnyOrder(2, 4, 6); p.run(); }
@Test public void testFilterParDoOutputTypeDescriptorRawWithLambda() throws Exception { @SuppressWarnings({"unchecked", "rawtypes"}) PCollection<String> output = p.apply(Create.of("hello")).apply(Filter.by(s -> true)); thrown.expect(CannotProvideCoderException.class); p.getCoderRegistry().getCoder(output.getTypeDescriptor()); }
@Test @Category(NeedsRunner.class) public void testFilterByMethodReferenceWithLambda() { PCollection<Integer> output = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7)).apply(Filter.by(new EvenFilter()::isEven)); PAssert.that(output).containsInAnyOrder(2, 4, 6); p.run(); }