Group { public static <T> Global<T> globally() { return new Global<>(); }  static Global<T> globally(); static ByFields<T> byFieldNames(String... fieldNames); static ByFields<T> byFieldNames(Iterable<String> fieldNames); static ByFields<T> byFieldIds(Integer... fieldIds); static ByFields<T> byFieldIds(Iterable<Integer> fieldIds); static ByFields<T> byFieldAccessDescriptor(FieldAccessDescriptor fieldAccess);  }
@Test @Category(NeedsRunner.class) public void testGroupGlobally() { Collection<Basic> elements = ImmutableList.of( Basic.of("key1", 1, "value1"), Basic.of("key1", 1, "value2"), Basic.of("key2", 2, "value3"), Basic.of("key2", 2, "value4")); PCollection<Iterable<Basic>> grouped = pipeline.apply(Create.of(elements)).apply(Group.globally()); PAssert.that(grouped).satisfies(actual -> containsSingleIterable(elements, actual)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGlobalAggregation() { Collection<Basic> elements = ImmutableList.of( Basic.of("key1", 1, "value1"), Basic.of("key1", 1, "value2"), Basic.of("key2", 2, "value3"), Basic.of("key2", 2, "value4")); PCollection<Long> count = pipeline .apply(Create.of(elements)) .apply(Group.<Basic>globally().aggregate(Count.combineFn())); PAssert.that(count).containsInAnyOrder(4L); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGloballyWithSchemaAggregateFn() { Collection<Aggregate> elements = ImmutableList.of( Aggregate.of(1, 1, 2), Aggregate.of(2, 1, 3), Aggregate.of(3, 2, 4), Aggregate.of(4, 2, 5)); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<Aggregate>globally() .aggregateField("field1", Sum.ofLongs(), "field1_sum") .aggregateField("field3", Sum.ofIntegers(), "field3_sum") .aggregateField("field1", Top.largestLongsFn(1), "field1_top")); Schema aggregateSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Row expectedRow = Row.withSchema(aggregateSchema).addValues(10L, 14).addArray(4L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testAggregateByMultipleFields() { Collection<Aggregate> elements = ImmutableList.of( Aggregate.of(1, 1, 2), Aggregate.of(2, 1, 3), Aggregate.of(3, 2, 4), Aggregate.of(4, 2, 5)); List<String> fieldNames = Lists.newArrayList("field1", "field2"); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<Aggregate>globally() .aggregateFields(fieldNames, new MultipleFieldCombineFn(), "field1+field2")); Schema outputSchema = Schema.builder().addInt64Field("field1+field2").build(); Row expectedRow = Row.withSchema(outputSchema).addValues(16L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGloballyWithSchemaAggregateFnNestedFields() { Collection<OuterAggregate> elements = ImmutableList.of( OuterAggregate.of(Aggregate.of(1, 1, 2)), OuterAggregate.of(Aggregate.of(2, 1, 3)), OuterAggregate.of(Aggregate.of(3, 2, 4)), OuterAggregate.of(Aggregate.of(4, 2, 5))); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<OuterAggregate>globally() .aggregateField("inner.field1", Sum.ofLongs(), "field1_sum") .aggregateField("inner.field3", Sum.ofIntegers(), "field3_sum") .aggregateField("inner.field1", Top.largestLongsFn(1), "field1_top")); Schema aggregateSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Row expectedRow = Row.withSchema(aggregateSchema).addValues(10L, 14).addArray(4L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testAggregateBaseValuesGlobally() { Collection<BasicEnum> elements = Lists.newArrayList( BasicEnum.of("a", BasicEnum.Test.ONE), BasicEnum.of("a", BasicEnum.Test.TWO)); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<BasicEnum>globally() .aggregateFieldBaseValue("enumeration", Sum.ofIntegers(), "enum_sum")); Schema aggregateSchema = Schema.builder().addInt32Field("enum_sum").build(); Row expectedRow = Row.withSchema(aggregateSchema).addValues(3).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testAggregateLogicalValuesGlobally() { Collection<BasicEnum> elements = Lists.newArrayList( BasicEnum.of("a", BasicEnum.Test.ONE), BasicEnum.of("a", BasicEnum.Test.TWO)); CombineFn<EnumerationType.Value, ?, Iterable<EnumerationType.Value>> sampleAnyCombineFn = Sample.anyCombineFn(100); Field aggField = Field.of("sampleList", FieldType.array(FieldType.logicalType(BASIC_ENUM_ENUMERATION))); pipeline .apply(Create.of(elements)) .apply( Group.<BasicEnum>globally().aggregateField("enumeration", sampleAnyCombineFn, aggField)) .apply( ParDo.of( new DoFn<Row, List<Integer>>() { @ProcessElement public void process(@Element Row value) { assertThat( value.getArray(0), containsInAnyOrder( BASIC_ENUM_ENUMERATION.valueOf(1), BASIC_ENUM_ENUMERATION.valueOf(2))); } })); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupGlobally() { Collection<POJO> elements = ImmutableList.of( new POJO("key1", 1, "value1"), new POJO("key1", 1, "value2"), new POJO("key2", 2, "value3"), new POJO("key2", 2, "value4")); PCollection<Iterable<POJO>> grouped = pipeline.apply(Create.of(elements)).apply(Group.globally()); PAssert.that(grouped).satisfies(actual -> containsSingleIterable(elements, actual)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGlobalAggregation() { Collection<POJO> elements = ImmutableList.of( new POJO("key1", 1, "value1"), new POJO("key1", 1, "value2"), new POJO("key2", 2, "value3"), new POJO("key2", 2, "value4")); PCollection<Long> count = pipeline .apply(Create.of(elements)) .apply(Group.<POJO>globally().aggregate(Count.combineFn())); PAssert.that(count).containsInAnyOrder(4L); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGloballyWithSchemaAggregateFn() { Collection<AggregatePojos> elements = ImmutableList.of( new AggregatePojos(1, 1, 2), new AggregatePojos(2, 1, 3), new AggregatePojos(3, 2, 4), new AggregatePojos(4, 2, 5)); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<AggregatePojos>globally() .aggregateField("field1", Sum.ofLongs(), "field1_sum") .aggregateField("field3", Sum.ofIntegers(), "field3_sum") .aggregateField("field1", Top.largestLongsFn(1), "field1_top")); Schema aggregateSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Row expectedRow = Row.withSchema(aggregateSchema).addValues(10L, 14).addArray(4L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testAggregateByMultipleFields() { Collection<AggregatePojos> elements = ImmutableList.of( new AggregatePojos(1, 1, 2), new AggregatePojos(2, 1, 3), new AggregatePojos(3, 2, 4), new AggregatePojos(4, 2, 5)); List<String> fieldNames = Lists.newArrayList("field1", "field2"); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<AggregatePojos>globally() .aggregateFields(fieldNames, new MultipleFieldCombineFn(), "field1+field2")); Schema outputSchema = Schema.builder().addInt64Field("field1+field2").build(); Row expectedRow = Row.withSchema(outputSchema).addValues(16L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGloballyWithSchemaAggregateFnNestedFields() { Collection<OuterAggregate> elements = ImmutableList.of( new OuterAggregate(new AggregatePojos(1, 1, 2)), new OuterAggregate(new AggregatePojos(2, 1, 3)), new OuterAggregate(new AggregatePojos(3, 2, 4)), new OuterAggregate(new AggregatePojos(4, 2, 5))); PCollection<Row> aggregate = pipeline .apply(Create.of(elements)) .apply( Group.<OuterAggregate>globally() .aggregateField("inner.field1", Sum.ofLongs(), "field1_sum") .aggregateField("inner.field3", Sum.ofIntegers(), "field3_sum") .aggregateField("inner.field1", Top.largestLongsFn(1), "field1_top")); Schema aggregateSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Row expectedRow = Row.withSchema(aggregateSchema).addValues(10L, 14).addArray(4L).build(); PAssert.that(aggregate).containsInAnyOrder(expectedRow); pipeline.run(); }