Group { public static <T> ByFields<T> byFieldNames(String... fieldNames) { return ByFields.of(FieldAccessDescriptor.withFieldNames(fieldNames)); }  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 testGroupByOneField() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( Basic.of("key1", 1, "value1"), Basic.of("key1", 2, "value2"), Basic.of("key2", 3, "value3"), Basic.of("key2", 4, "value4"))) .apply(Group.byFieldNames("field1")); Schema keySchema = Schema.builder().addStringField("field1").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(BASIC_SCHEMA)) .build(); SerializableFunction<Basic, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(Basic.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue("key1").build()) .addIterable( ImmutableList.of( toRow.apply(Basic.of("key1", 1L, "value1")), toRow.apply(Basic.of("key1", 2L, "value2")))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue("key2").build()) .addIterable( ImmutableList.of( toRow.apply(Basic.of("key2", 3L, "value3")), toRow.apply(Basic.of("key2", 4L, "value4")))) .build()); PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual, new Basic[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupByMultiple() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( Basic.of("key1", 1, "value1"), Basic.of("key1", 1, "value2"), Basic.of("key2", 2, "value3"), Basic.of("key2", 2, "value4"))) .apply(Group.byFieldNames("field1", "field2")); Schema keySchema = Schema.builder().addStringField("field1").addInt64Field("field2").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(BASIC_SCHEMA)) .build(); SerializableFunction<Basic, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(Basic.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key1", 1L).build()) .addIterable( ImmutableList.of( toRow.apply(Basic.of("key1", 1L, "value1")), toRow.apply(Basic.of("key1", 1L, "value2")))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key2", 2L).build()) .addIterable( ImmutableList.of( toRow.apply(Basic.of("key2", 2L, "value3")), toRow.apply(Basic.of("key2", 2L, "value4")))) .build()); PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual, new Basic[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupByNestedKey() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( Outer.of(Basic.of("key1", 1L, "value1")), Outer.of(Basic.of("key1", 1L, "value2")), Outer.of(Basic.of("key2", 2L, "value3")), Outer.of(Basic.of("key2", 2L, "value4")))) .apply(Group.byFieldNames("inner.field1", "inner.field2")); Schema keySchema = Schema.builder().addStringField("field1").addInt64Field("field2").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(OUTER_SCHEMA)) .build(); SerializableFunction<Outer, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(Outer.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key1", 1L).build()) .addIterable( ImmutableList.of( toRow.apply(Outer.of(Basic.of("key1", 1L, "value1"))), toRow.apply(Outer.of(Basic.of("key1", 1L, "value2"))))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key2", 2L).build()) .addIterable( ImmutableList.of( toRow.apply(Outer.of(Basic.of("key2", 2L, "value3"))), toRow.apply(Outer.of(Basic.of("key2", 2L, "value4"))))) .build()); PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual, new Outer[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testOutputCoders() { Schema keySchema = Schema.builder().addStringField("field1").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(BASIC_SCHEMA)) .build(); PCollection<Row> grouped = pipeline .apply(Create.of(Basic.of("key1", 1, "value1"))) .apply(Group.byFieldNames("field1")); assertTrue(grouped.getSchema().equivalent(outputSchema)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testByKeyWithSchemaAggregateFn() { 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> aggregations = pipeline .apply(Create.of(elements)) .apply( Group.<Aggregate>byFieldNames("field2") .aggregateField("field1", Sum.ofLongs(), "field1_sum") .aggregateField("field3", Sum.ofIntegers(), "field3_sum") .aggregateField("field1", Top.largestLongsFn(1), "field1_top")); Schema keySchema = Schema.builder().addInt64Field("field2").build(); Schema valueSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Schema outputSchema = Schema.builder().addRowField("key", keySchema).addRowField("value", valueSchema).build(); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(1L).build()) .addValue(Row.withSchema(valueSchema).addValue(3L).addValue(5).addArray(2L).build()) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(2L).build()) .addValue(Row.withSchema(valueSchema).addValue(7L).addValue(9).addArray(4L).build()) .build()); PAssert.that(aggregations).satisfies(actual -> containsKvRows(expected, actual)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testByKeyWithSchemaAggregateFnNestedFields() { 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> aggregations = pipeline .apply(Create.of(elements)) .apply( Group.<OuterAggregate>byFieldNames("inner.field2") .aggregateField("inner.field1", Sum.ofLongs(), "field1_sum") .aggregateField("inner.field3", Sum.ofIntegers(), "field3_sum") .aggregateField("inner.field1", Top.largestLongsFn(1), "field1_top")); Schema keySchema = Schema.builder().addInt64Field("field2").build(); Schema valueSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Schema outputSchema = Schema.builder().addRowField("key", keySchema).addRowField("value", valueSchema).build(); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(1L).build()) .addValue(Row.withSchema(valueSchema).addValue(3L).addValue(5).addArray(2L).build()) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(2L).build()) .addValue(Row.withSchema(valueSchema).addValue(7L).addValue(9).addArray(4L).build()) .build()); PAssert.that(aggregations).satisfies(actual -> containsKvRows(expected, actual)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupByOneField() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( new POJO("key1", 1, "value1"), new POJO("key1", 2, "value2"), new POJO("key2", 3, "value3"), new POJO("key2", 4, "value4"))) .apply(Group.byFieldNames("field1")); Schema keySchema = Schema.builder().addStringField("field1").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(POJO_SCHEMA)) .build(); SerializableFunction<POJO, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(POJO.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue("key1").build()) .addIterable( ImmutableList.of( toRow.apply(new POJO("key1", 1L, "value1")), toRow.apply(new POJO("key1", 2L, "value2")))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue("key2").build()) .addIterable( ImmutableList.of( toRow.apply(new POJO("key2", 3L, "value3")), toRow.apply(new POJO("key2", 4L, "value4")))) .build()); PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual, new POJO[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupByMultiple() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( new POJO("key1", 1, "value1"), new POJO("key1", 1, "value2"), new POJO("key2", 2, "value3"), new POJO("key2", 2, "value4"))) .apply(Group.byFieldNames("field1", "field2")); Schema keySchema = Schema.builder().addStringField("field1").addInt64Field("field2").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(POJO_SCHEMA)) .build(); SerializableFunction<POJO, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(POJO.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key1", 1L).build()) .addIterable( ImmutableList.of( toRow.apply(new POJO("key1", 1L, "value1")), toRow.apply(new POJO("key1", 1L, "value2")))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key2", 2L).build()) .addIterable( ImmutableList.of( toRow.apply(new POJO("key2", 2L, "value3")), toRow.apply(new POJO("key2", 2L, "value4")))) .build()); PAssert.that(grouped).satisfies(actual -> containsKIterableVs(expected, actual, new POJO[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testGroupByNestedKey() throws NoSuchSchemaException { PCollection<Row> grouped = pipeline .apply( Create.of( new OuterPOJO(new POJO("key1", 1L, "value1")), new OuterPOJO(new POJO("key1", 1L, "value2")), new OuterPOJO(new POJO("key2", 2L, "value3")), new OuterPOJO(new POJO("key2", 2L, "value4")))) .apply(Group.byFieldNames("inner.field1", "inner.field2")); Schema keySchema = Schema.builder().addStringField("field1").addInt64Field("field2").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(OUTER_POJO_SCHEMA)) .build(); SerializableFunction<OuterPOJO, Row> toRow = pipeline.getSchemaRegistry().getToRowFunction(OuterPOJO.class); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key1", 1L).build()) .addIterable( ImmutableList.of( toRow.apply(new OuterPOJO(new POJO("key1", 1L, "value1"))), toRow.apply(new OuterPOJO(new POJO("key1", 1L, "value2"))))) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValues("key2", 2L).build()) .addIterable( ImmutableList.of( toRow.apply(new OuterPOJO(new POJO("key2", 2L, "value3"))), toRow.apply(new OuterPOJO(new POJO("key2", 2L, "value4"))))) .build()); PAssert.that(grouped) .satisfies(actual -> containsKIterableVs(expected, actual, new OuterPOJO[0])); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testOutputCoders() { Schema keySchema = Schema.builder().addStringField("field1").build(); Schema outputSchema = Schema.builder() .addRowField("key", keySchema) .addIterableField("value", FieldType.row(POJO_SCHEMA)) .build(); PCollection<Row> grouped = pipeline .apply(Create.of(new POJO("key1", 1, "value1"))) .apply(Group.byFieldNames("field1")); assertTrue(grouped.getSchema().equivalent(outputSchema)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testByKeyWithSchemaAggregateFn() { 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> aggregations = pipeline .apply(Create.of(elements)) .apply( Group.<AggregatePojos>byFieldNames("field2") .aggregateField("field1", Sum.ofLongs(), "field1_sum") .aggregateField("field3", Sum.ofIntegers(), "field3_sum") .aggregateField("field1", Top.largestLongsFn(1), "field1_top")); Schema keySchema = Schema.builder().addInt64Field("field2").build(); Schema valueSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Schema outputSchema = Schema.builder().addRowField("key", keySchema).addRowField("value", valueSchema).build(); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(1L).build()) .addValue(Row.withSchema(valueSchema).addValue(3L).addValue(5).addArray(2L).build()) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(2L).build()) .addValue(Row.withSchema(valueSchema).addValue(7L).addValue(9).addArray(4L).build()) .build()); PAssert.that(aggregations).satisfies(actual -> containsKvRows(expected, actual)); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testByKeyWithSchemaAggregateFnNestedFields() { 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> aggregations = pipeline .apply(Create.of(elements)) .apply( Group.<OuterAggregate>byFieldNames("inner.field2") .aggregateField("inner.field1", Sum.ofLongs(), "field1_sum") .aggregateField("inner.field3", Sum.ofIntegers(), "field3_sum") .aggregateField("inner.field1", Top.largestLongsFn(1), "field1_top")); Schema keySchema = Schema.builder().addInt64Field("field2").build(); Schema valueSchema = Schema.builder() .addInt64Field("field1_sum") .addInt32Field("field3_sum") .addArrayField("field1_top", FieldType.INT64) .build(); Schema outputSchema = Schema.builder().addRowField("key", keySchema).addRowField("value", valueSchema).build(); List<Row> expected = ImmutableList.of( Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(1L).build()) .addValue(Row.withSchema(valueSchema).addValue(3L).addValue(5).addArray(2L).build()) .build(), Row.withSchema(outputSchema) .addValue(Row.withSchema(keySchema).addValue(2L).build()) .addValue(Row.withSchema(valueSchema).addValue(7L).addValue(9).addArray(4L).build()) .build()); PAssert.that(aggregations).satisfies(actual -> containsKvRows(expected, actual)); pipeline.run(); }