@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(); } }
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()))); }
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()))); } }
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()))); }  }
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); }
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);  }