### Question:
Aggregate { public static <K, V> PTable<K, Collection<V>> collectValues(PTable<K, V> collect) { PTypeFamily tf = collect.getTypeFamily(); final PType<V> valueType = collect.getValueType(); return collect.groupByKey().parallelDo("collect", new MapValuesFn<K, Iterable<V>, Collection<V>>() { public Collection<V> map(Iterable<V> values) { List<V> collected = Lists.newArrayList(); for (V value : values) { collected.add(valueType.getDetachedValue(value)); } return collected; } }, tf.tableOf(collect.getKeyType(), tf.collections(collect.getValueType()))); }  static PTable<S, Long> count(PCollection<S> collect); static PTable<K, V> top(PTable<K, V> ptable, int limit, boolean maximize); static PCollection<S> max(PCollection<S> collect); static PCollection<S> min(PCollection<S> collect); static PTable<K, Collection<V>> collectValues(PTable<K, V> collect);  }### Answer:
@Test public void testCollectUrls() throws Exception { Pipeline p = new MRPipeline(AggregateTest.class); String urlsInputPath = FileHelper.createTempCopyOf("urls.txt"); PTable<String, Collection<String>> urls = Aggregate.collectValues( p.readTextFile(urlsInputPath) .parallelDo(new SplitFn(), tableOf(strings(), strings()))); for (Pair<String, Collection<String>> e : urls.materialize()) { String key = e.first(); int expectedSize = 0; if ("www.A.com".equals(key)) { expectedSize = 4; } else if ("www.B.com".equals(key) || "www.F.com".equals(key)) { expectedSize = 2; } else if ("www.C.com".equals(key) || "www.D.com".equals(key) || "www.E.com".equals(key)) { expectedSize = 1; } assertEquals("Checking key = " + key, expectedSize, e.second().size()); p.done(); } }
@Test public void testCollectValues_Writables() throws IOException { Pipeline pipeline = new MRPipeline(AggregateTest.class); Map<Integer, Collection<Text>> collectionMap = pipeline .readTextFile(FileHelper.createTempCopyOf("set2.txt")) .parallelDo(new MapStringToTextPair(), Writables.tableOf(Writables.ints(), Writables.writables(Text.class)) ).collectValues().materializeToMap(); assertEquals(1, collectionMap.size()); assertEquals(Lists.newArrayList(new Text("c"), new Text("d"), new Text("a")), collectionMap.get(1)); }
@Test public void testCollectValues_Avro() throws IOException { MapStringToEmployeePair mapFn = new MapStringToEmployeePair(); Pipeline pipeline = new MRPipeline(AggregateTest.class); Map<Integer, Collection<Employee>> collectionMap = pipeline .readTextFile(FileHelper.createTempCopyOf("set2.txt")) .parallelDo(mapFn, Avros.tableOf(Avros.ints(), Avros.records(Employee.class))).collectValues() .materializeToMap(); assertEquals(1, collectionMap.size()); Employee empC = mapFn.map("c").second(); Employee empD = mapFn.map("d").second(); Employee empA = mapFn.map("a").second(); assertEquals(Lists.newArrayList(empC, empD, empA), collectionMap.get(1)); }