You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/03/24 03:47:43 UTC
[19/67] [partial] incubator-beam git commit: Directory reorganization
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Sum.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Sum.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Sum.java
deleted file mode 100644
index 5b30475..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Sum.java
+++ /dev/null
@@ -1,188 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import com.google.cloud.dataflow.sdk.util.common.Counter;
-import com.google.cloud.dataflow.sdk.util.common.Counter.AggregationKind;
-import com.google.cloud.dataflow.sdk.util.common.CounterProvider;
-
-/**
- * {@code PTransform}s for computing the sum of the elements in a
- * {@code PCollection}, or the sum of the values associated with
- * each key in a {@code PCollection} of {@code KV}s.
- *
- * <p>Example 1: get the sum of a {@code PCollection} of {@code Double}s.
- * <pre> {@code
- * PCollection<Double> input = ...;
- * PCollection<Double> sum = input.apply(Sum.doublesGlobally());
- * } </pre>
- *
- * <p>Example 2: calculate the sum of the {@code Integer}s
- * associated with each unique key (which is of type {@code String}).
- * <pre> {@code
- * PCollection<KV<String, Integer>> input = ...;
- * PCollection<KV<String, Integer>> sumPerKey = input
- * .apply(Sum.<String>integersPerKey());
- * } </pre>
- */
-public class Sum {
-
- private Sum() {
- // do not instantiate
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<Integer>} and returns a
- * {@code PCollection<Integer>} whose contents is the sum of the
- * input {@code PCollection}'s elements, or
- * {@code 0} if there are no elements.
- */
- public static Combine.Globally<Integer, Integer> integersGlobally() {
- return Combine.globally(new SumIntegerFn()).named("Sum.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, Integer>>} and returns a
- * {@code PCollection<KV<K, Integer>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the sum of the values associated with
- * that key in the input {@code PCollection}.
- */
- public static <K> Combine.PerKey<K, Integer, Integer> integersPerKey() {
- return Combine.<K, Integer, Integer>perKey(new SumIntegerFn()).named("Sum.PerKey");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<Long>} and returns a
- * {@code PCollection<Long>} whose contents is the sum of the
- * input {@code PCollection}'s elements, or
- * {@code 0} if there are no elements.
- */
- public static Combine.Globally<Long, Long> longsGlobally() {
- return Combine.globally(new SumLongFn()).named("Sum.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, Long>>} and returns a
- * {@code PCollection<KV<K, Long>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the sum of the values associated with
- * that key in the input {@code PCollection}.
- */
- public static <K> Combine.PerKey<K, Long, Long> longsPerKey() {
- return Combine.<K, Long, Long>perKey(new SumLongFn()).named("Sum.PerKey");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<Double>} and returns a
- * {@code PCollection<Double>} whose contents is the sum of the
- * input {@code PCollection}'s elements, or
- * {@code 0} if there are no elements.
- */
- public static Combine.Globally<Double, Double> doublesGlobally() {
- return Combine.globally(new SumDoubleFn()).named("Sum.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, Double>>} and returns a
- * {@code PCollection<KV<K, Double>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the sum of the values associated with
- * that key in the input {@code PCollection}.
- */
- public static <K> Combine.PerKey<K, Double, Double> doublesPerKey() {
- return Combine.<K, Double, Double>perKey(new SumDoubleFn()).named("Sum.PerKey");
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * A {@code SerializableFunction} that computes the sum of an
- * {@code Iterable} of {@code Integer}s, useful as an argument to
- * {@link Combine#globally} or {@link Combine#perKey}.
- */
- public static class SumIntegerFn
- extends Combine.BinaryCombineIntegerFn implements CounterProvider<Integer> {
- @Override
- public int apply(int a, int b) {
- return a + b;
- }
-
- @Override
- public int identity() {
- return 0;
- }
-
- @Override
- public Counter<Integer> getCounter(String name) {
- return Counter.ints(name, AggregationKind.SUM);
- }
- }
-
- /**
- * A {@code SerializableFunction} that computes the sum of an
- * {@code Iterable} of {@code Long}s, useful as an argument to
- * {@link Combine#globally} or {@link Combine#perKey}.
- */
- public static class SumLongFn
- extends Combine.BinaryCombineLongFn implements CounterProvider<Long> {
- @Override
- public long apply(long a, long b) {
- return a + b;
- }
-
- @Override
- public long identity() {
- return 0;
- }
-
- @Override
- public Counter<Long> getCounter(String name) {
- return Counter.longs(name, AggregationKind.SUM);
- }
- }
-
- /**
- * A {@code SerializableFunction} that computes the sum of an
- * {@code Iterable} of {@code Double}s, useful as an argument to
- * {@link Combine#globally} or {@link Combine#perKey}.
- */
- public static class SumDoubleFn
- extends Combine.BinaryCombineDoubleFn implements CounterProvider<Double> {
- @Override
- public double apply(double a, double b) {
- return a + b;
- }
-
- @Override
- public double identity() {
- return 0;
- }
-
- @Override
- public Counter<Double> getCounter(String name) {
- return Counter.doubles(name, AggregationKind.SUM);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Top.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Top.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Top.java
deleted file mode 100644
index 98fe53c..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Top.java
+++ /dev/null
@@ -1,559 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.CoderException;
-import com.google.cloud.dataflow.sdk.coders.CoderRegistry;
-import com.google.cloud.dataflow.sdk.coders.CustomCoder;
-import com.google.cloud.dataflow.sdk.coders.ListCoder;
-import com.google.cloud.dataflow.sdk.transforms.Combine.AccumulatingCombineFn;
-import com.google.cloud.dataflow.sdk.transforms.Combine.AccumulatingCombineFn.Accumulator;
-import com.google.cloud.dataflow.sdk.transforms.Combine.PerKey;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObserver;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.List;
-import java.util.PriorityQueue;
-
-/**
- * {@code PTransform}s for finding the largest (or smallest) set
- * of elements in a {@code PCollection}, or the largest (or smallest)
- * set of values associated with each key in a {@code PCollection} of
- * {@code KV}s.
- */
-public class Top {
-
- private Top() {
- // do not instantiate
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<T>} and returns a {@code PCollection<List<T>>} with a
- * single element containing the largest {@code count} elements of the input
- * {@code PCollection<T>}, in decreasing order, sorted using the
- * given {@code Comparator<T>}. The {@code Comparator<T>} must also
- * be {@code Serializable}.
- *
- * <p>If {@code count} {@code <} the number of elements in the
- * input {@code PCollection}, then all the elements of the input
- * {@code PCollection} will be in the resulting
- * {@code List}, albeit in sorted order.
- *
- * <p>All the elements of the result's {@code List}
- * must fit into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<Student> students = ...;
- * PCollection<List<Student>> top10Students =
- * students.apply(Top.of(10, new CompareStudentsByAvgGrade()));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the output {@code PCollection}
- * is a {@code ListCoder} of the {@code Coder} of the elements of
- * the input {@code PCollection}.
- *
- * <p>If the input {@code PCollection} is windowed into {@link GlobalWindows},
- * an empty {@code List<T>} in the {@link GlobalWindow} will be output if the input
- * {@code PCollection} is empty. To use this with inputs with other windowing,
- * either {@link Combine.Globally#withoutDefaults withoutDefaults} or
- * {@link Combine.Globally#asSingletonView asSingletonView} must be called.
- *
- * <p>See also {@link #smallest} and {@link #largest}, which sort
- * {@code Comparable} elements using their natural ordering.
- *
- * <p>See also {@link #perKey}, {@link #smallestPerKey}, and
- * {@link #largestPerKey}, which take a {@code PCollection} of
- * {@code KV}s and return the top values associated with each key.
- */
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- Combine.Globally<T, List<T>> of(int count, ComparatorT compareFn) {
- return Combine.globally(new TopCombineFn<>(count, compareFn)).named("Top.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<T>} and returns a {@code PCollection<List<T>>} with a
- * single element containing the smallest {@code count} elements of the input
- * {@code PCollection<T>}, in increasing order, sorted according to
- * their natural order.
- *
- * <p>If {@code count} {@code <} the number of elements in the
- * input {@code PCollection}, then all the elements of the input
- * {@code PCollection} will be in the resulting {@code PCollection}'s
- * {@code List}, albeit in sorted order.
- *
- * <p>All the elements of the result {@code List}
- * must fit into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<Integer> values = ...;
- * PCollection<List<Integer>> smallest10Values = values.apply(Top.smallest(10));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the output {@code PCollection}
- * is a {@code ListCoder} of the {@code Coder} of the elements of
- * the input {@code PCollection}.
- *
- * <p>If the input {@code PCollection} is windowed into {@link GlobalWindows},
- * an empty {@code List<T>} in the {@link GlobalWindow} will be output if the input
- * {@code PCollection} is empty. To use this with inputs with other windowing,
- * either {@link Combine.Globally#withoutDefaults withoutDefaults} or
- * {@link Combine.Globally#asSingletonView asSingletonView} must be called.
- *
- * <p>See also {@link #largest}.
- *
- * <p>See also {@link #of}, which sorts using a user-specified
- * {@code Comparator} function.
- *
- * <p>See also {@link #perKey}, {@link #smallestPerKey}, and
- * {@link #largestPerKey}, which take a {@code PCollection} of
- * {@code KV}s and return the top values associated with each key.
- */
- public static <T extends Comparable<T>> Combine.Globally<T, List<T>> smallest(int count) {
- return Combine.globally(new TopCombineFn<>(count, new Smallest<T>()))
- .named("Smallest.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<T>} and returns a {@code PCollection<List<T>>} with a
- * single element containing the largest {@code count} elements of the input
- * {@code PCollection<T>}, in decreasing order, sorted according to
- * their natural order.
- *
- * <p>If {@code count} {@code <} the number of elements in the
- * input {@code PCollection}, then all the elements of the input
- * {@code PCollection} will be in the resulting {@code PCollection}'s
- * {@code List}, albeit in sorted order.
- *
- * <p>All the elements of the result's {@code List}
- * must fit into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<Integer> values = ...;
- * PCollection<List<Integer>> largest10Values = values.apply(Top.largest(10));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the output {@code PCollection}
- * is a {@code ListCoder} of the {@code Coder} of the elements of
- * the input {@code PCollection}.
- *
- * <p>If the input {@code PCollection} is windowed into {@link GlobalWindows},
- * an empty {@code List<T>} in the {@link GlobalWindow} will be output if the input
- * {@code PCollection} is empty. To use this with inputs with other windowing,
- * either {@link Combine.Globally#withoutDefaults withoutDefaults} or
- * {@link Combine.Globally#asSingletonView asSingletonView} must be called.
- *
- * <p>See also {@link #smallest}.
- *
- * <p>See also {@link #of}, which sorts using a user-specified
- * {@code Comparator} function.
- *
- * <p>See also {@link #perKey}, {@link #smallestPerKey}, and
- * {@link #largestPerKey}, which take a {@code PCollection} of
- * {@code KV}s and return the top values associated with each key.
- */
- public static <T extends Comparable<T>> Combine.Globally<T, List<T>> largest(int count) {
- return Combine.globally(new TopCombineFn<>(count, new Largest<T>())).named("Largest.Globally");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, V>>} and returns a
- * {@code PCollection<KV<K, List<V>>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the largest {@code count} values
- * associated with that key in the input
- * {@code PCollection<KV<K, V>>}, in decreasing order, sorted using
- * the given {@code Comparator<V>}. The
- * {@code Comparator<V>} must also be {@code Serializable}.
- *
- * <p>If there are fewer than {@code count} values associated with
- * a particular key, then all those values will be in the result
- * mapping for that key, albeit in sorted order.
- *
- * <p>All the values associated with a single key must fit into the
- * memory of a single machine, but there can be many more
- * {@code KV}s in the resulting {@code PCollection} than can fit
- * into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<School, Student>> studentsBySchool = ...;
- * PCollection<KV<School, List<Student>>> top10StudentsBySchool =
- * studentsBySchool.apply(
- * Top.perKey(10, new CompareStudentsByAvgGrade()));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the keys of the output
- * {@code PCollection} is the same as that of the keys of the input
- * {@code PCollection}, and the {@code Coder} of the values of the
- * output {@code PCollection} is a {@code ListCoder} of the
- * {@code Coder} of the values of the input {@code PCollection}.
- *
- * <p>See also {@link #smallestPerKey} and {@link #largestPerKey}, which
- * sort {@code Comparable<V>} values using their natural
- * ordering.
- *
- * <p>See also {@link #of}, {@link #smallest}, and {@link #largest}, which
- * take a {@code PCollection} and return the top elements.
- */
- public static <K, V, ComparatorT extends Comparator<V> & Serializable>
- PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>>
- perKey(int count, ComparatorT compareFn) {
- return Combine.perKey(
- new TopCombineFn<>(count, compareFn).<K>asKeyedFn()).named("Top.PerKey");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, V>>} and returns a
- * {@code PCollection<KV<K, List<V>>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the smallest {@code count} values
- * associated with that key in the input
- * {@code PCollection<KV<K, V>>}, in increasing order, sorted
- * according to their natural order.
- *
- * <p>If there are fewer than {@code count} values associated with
- * a particular key, then all those values will be in the result
- * mapping for that key, albeit in sorted order.
- *
- * <p>All the values associated with a single key must fit into the
- * memory of a single machine, but there can be many more
- * {@code KV}s in the resulting {@code PCollection} than can fit
- * into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<String, Integer>> keyedValues = ...;
- * PCollection<KV<String, List<Integer>>> smallest10ValuesPerKey =
- * keyedValues.apply(Top.smallestPerKey(10));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the keys of the output
- * {@code PCollection} is the same as that of the keys of the input
- * {@code PCollection}, and the {@code Coder} of the values of the
- * output {@code PCollection} is a {@code ListCoder} of the
- * {@code Coder} of the values of the input {@code PCollection}.
- *
- * <p>See also {@link #largestPerKey}.
- *
- * <p>See also {@link #perKey}, which sorts values using a user-specified
- * {@code Comparator} function.
- *
- * <p>See also {@link #of}, {@link #smallest}, and {@link #largest}, which
- * take a {@code PCollection} and return the top elements.
- */
- public static <K, V extends Comparable<V>>
- PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>>
- smallestPerKey(int count) {
- return Combine.perKey(new TopCombineFn<>(count, new Smallest<V>()).<K>asKeyedFn())
- .named("Smallest.PerKey");
- }
-
- /**
- * Returns a {@code PTransform} that takes an input
- * {@code PCollection<KV<K, V>>} and returns a
- * {@code PCollection<KV<K, List<V>>>} that contains an output
- * element mapping each distinct key in the input
- * {@code PCollection} to the largest {@code count} values
- * associated with that key in the input
- * {@code PCollection<KV<K, V>>}, in decreasing order, sorted
- * according to their natural order.
- *
- * <p>If there are fewer than {@code count} values associated with
- * a particular key, then all those values will be in the result
- * mapping for that key, albeit in sorted order.
- *
- * <p>All the values associated with a single key must fit into the
- * memory of a single machine, but there can be many more
- * {@code KV}s in the resulting {@code PCollection} than can fit
- * into the memory of a single machine.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<String, Integer>> keyedValues = ...;
- * PCollection<KV<String, List<Integer>>> largest10ValuesPerKey =
- * keyedValues.apply(Top.largestPerKey(10));
- * } </pre>
- *
- * <p>By default, the {@code Coder} of the keys of the output
- * {@code PCollection} is the same as that of the keys of the input
- * {@code PCollection}, and the {@code Coder} of the values of the
- * output {@code PCollection} is a {@code ListCoder} of the
- * {@code Coder} of the values of the input {@code PCollection}.
- *
- * <p>See also {@link #smallestPerKey}.
- *
- * <p>See also {@link #perKey}, which sorts values using a user-specified
- * {@code Comparator} function.
- *
- * <p>See also {@link #of}, {@link #smallest}, and {@link #largest}, which
- * take a {@code PCollection} and return the top elements.
- */
- public static <K, V extends Comparable<V>>
- PerKey<K, V, List<V>>
- largestPerKey(int count) {
- return Combine.perKey(
-new TopCombineFn<>(count, new Largest<V>()).<K>asKeyedFn())
- .named("Largest.PerKey");
- }
-
- /**
- * A {@code Serializable} {@code Comparator} that that uses the compared elements' natural
- * ordering.
- */
- public static class Largest<T extends Comparable<? super T>>
- implements Comparator<T>, Serializable {
- @Override
- public int compare(T a, T b) {
- return a.compareTo(b);
- }
- }
-
- /**
- * {@code Serializable} {@code Comparator} that that uses the reverse of the compared elements'
- * natural ordering.
- */
- public static class Smallest<T extends Comparable<? super T>>
- implements Comparator<T>, Serializable {
- @Override
- public int compare(T a, T b) {
- return b.compareTo(a);
- }
- }
-
-
- ////////////////////////////////////////////////////////////////////////////
-
- /**
- * {@code CombineFn} for {@code Top} transforms that combines a
- * bunch of {@code T}s into a single {@code count}-long
- * {@code List<T>}, using {@code compareFn} to choose the largest
- * {@code T}s.
- *
- * @param <T> type of element being compared
- */
- public static class TopCombineFn<T, ComparatorT extends Comparator<T> & Serializable>
- extends AccumulatingCombineFn<T, BoundedHeap<T, ComparatorT>, List<T>> {
-
- private final int count;
- private final ComparatorT compareFn;
-
- public TopCombineFn(int count, ComparatorT compareFn) {
- Preconditions.checkArgument(
- count >= 0,
- "count must be >= 0");
- this.count = count;
- this.compareFn = compareFn;
- }
-
- @Override
- public BoundedHeap<T, ComparatorT> createAccumulator() {
- return new BoundedHeap<>(count, compareFn, new ArrayList<T>());
- }
-
- @Override
- public Coder<BoundedHeap<T, ComparatorT>> getAccumulatorCoder(
- CoderRegistry registry, Coder<T> inputCoder) {
- return new BoundedHeapCoder<>(count, compareFn, inputCoder);
- }
-
- @Override
- public String getIncompatibleGlobalWindowErrorMessage() {
- return "Default values are not supported in Top.[of, smallest, largest]() if the output "
- + "PCollection is not windowed by GlobalWindows. Instead, use "
- + "Top.[of, smallest, largest]().withoutDefaults() to output an empty PCollection if the"
- + " input PCollection is empty, or Top.[of, smallest, largest]().asSingletonView() to "
- + "get a PCollection containing the empty list if the input PCollection is empty.";
- }
- }
-
- /**
- * A heap that stores only a finite number of top elements according to its provided
- * {@code Comparator}. Implemented as an {@link Accumulator} to facilitate implementation of
- * {@link Top}.
- *
- * <p>This class is <i>not</i> safe for multithreaded use, except read-only.
- */
- static class BoundedHeap<T, ComparatorT extends Comparator<T> & Serializable>
- implements Accumulator<T, BoundedHeap<T, ComparatorT>, List<T>> {
-
- /**
- * A queue with smallest at the head, for quick adds.
- *
- * <p>Only one of asList and asQueue may be non-null.
- */
- private PriorityQueue<T> asQueue;
-
- /**
- * A list in with largest first, the form of extractOutput().
- *
- * <p>Only one of asList and asQueue may be non-null.
- */
- private List<T> asList;
-
- /** The user-provided Comparator. */
- private final ComparatorT compareFn;
-
- /** The maximum size of the heap. */
- private final int maximumSize;
-
- /**
- * Creates a new heap with the provided size, comparator, and initial elements.
- */
- private BoundedHeap(int maximumSize, ComparatorT compareFn, List<T> asList) {
- this.maximumSize = maximumSize;
- this.asList = asList;
- this.compareFn = compareFn;
- }
-
- @Override
- public void addInput(T value) {
- maybeAddInput(value);
- }
-
- /**
- * Adds {@code value} to this heap if it is larger than any of the current elements.
- * Returns {@code true} if {@code value} was added.
- */
- private boolean maybeAddInput(T value) {
- if (maximumSize == 0) {
- // Don't add anything.
- return false;
- }
-
- // If asQueue == null, then this is the first add after the latest call to the
- // constructor or asList().
- if (asQueue == null) {
- asQueue = new PriorityQueue<>(maximumSize, compareFn);
- for (T item : asList) {
- asQueue.add(item);
- }
- asList = null;
- }
-
- if (asQueue.size() < maximumSize) {
- asQueue.add(value);
- return true;
- } else if (compareFn.compare(value, asQueue.peek()) > 0) {
- asQueue.poll();
- asQueue.add(value);
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public void mergeAccumulator(BoundedHeap<T, ComparatorT> accumulator) {
- for (T value : accumulator.asList()) {
- if (!maybeAddInput(value)) {
- // If this element of accumulator does not make the top N, neither
- // will the rest, which are all smaller.
- break;
- }
- }
- }
-
- @Override
- public List<T> extractOutput() {
- return asList();
- }
-
- /**
- * Returns the contents of this Heap as a List sorted largest-to-smallest.
- */
- private List<T> asList() {
- if (asList == null) {
- List<T> smallestFirstList = Lists.newArrayListWithCapacity(asQueue.size());
- while (!asQueue.isEmpty()) {
- smallestFirstList.add(asQueue.poll());
- }
- asList = Lists.reverse(smallestFirstList);
- asQueue = null;
- }
- return asList;
- }
- }
-
- /**
- * A {@link Coder} for {@link BoundedHeap}, using Java serialization via {@link CustomCoder}.
- */
- private static class BoundedHeapCoder<T, ComparatorT extends Comparator<T> & Serializable>
- extends CustomCoder<BoundedHeap<T, ComparatorT>> {
- private final Coder<List<T>> listCoder;
- private final ComparatorT compareFn;
- private final int maximumSize;
-
- public BoundedHeapCoder(int maximumSize, ComparatorT compareFn, Coder<T> elementCoder) {
- listCoder = ListCoder.of(elementCoder);
- this.compareFn = compareFn;
- this.maximumSize = maximumSize;
- }
-
- @Override
- public void encode(
- BoundedHeap<T, ComparatorT> value, OutputStream outStream, Context context)
- throws CoderException, IOException {
- listCoder.encode(value.asList(), outStream, context);
- }
-
- @Override
- public BoundedHeap<T, ComparatorT> decode(InputStream inStream, Coder.Context context)
- throws CoderException, IOException {
- return new BoundedHeap<>(maximumSize, compareFn, listCoder.decode(inStream, context));
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- verifyDeterministic(
- "HeapCoder requires a deterministic list coder", listCoder);
- }
-
- @Override
- public boolean isRegisterByteSizeObserverCheap(
- BoundedHeap<T, ComparatorT> value, Context context) {
- return listCoder.isRegisterByteSizeObserverCheap(
- value.asList(), context);
- }
-
- @Override
- public void registerByteSizeObserver(
- BoundedHeap<T, ComparatorT> value, ElementByteSizeObserver observer, Context context)
- throws Exception {
- listCoder.registerByteSizeObserver(value.asList(), observer, context);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Values.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Values.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Values.java
deleted file mode 100644
index d84bc77..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Values.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-
-/**
- * {@code Values<V>} takes a {@code PCollection} of {@code KV<K, V>}s and
- * returns a {@code PCollection<V>} of the values.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<String, Long>> wordCounts = ...;
- * PCollection<Long> counts = wordCounts.apply(Values.<String>create());
- * } </pre>
- *
- * <p>Each output element has the same timestamp and is in the same windows
- * as its corresponding input element, and the output {@code PCollection}
- * has the same
- * {@link com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn}
- * associated with it as the input.
- *
- * <p>See also {@link Keys}.
- *
- * @param <V> the type of the values in the input {@code PCollection},
- * and the type of the elements in the output {@code PCollection}
- */
-public class Values<V> extends PTransform<PCollection<? extends KV<?, V>>,
- PCollection<V>> {
- /**
- * Returns a {@code Values<V>} {@code PTransform}.
- *
- * @param <V> the type of the values in the input {@code PCollection},
- * and the type of the elements in the output {@code PCollection}
- */
- public static <V> Values<V> create() {
- return new Values<>();
- }
-
- private Values() { }
-
- @Override
- public PCollection<V> apply(PCollection<? extends KV<?, V>> in) {
- return
- in.apply(ParDo.named("Values")
- .of(new DoFn<KV<?, V>, V>() {
- @Override
- public void processElement(ProcessContext c) {
- c.output(c.element().getValue());
- }
- }));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/View.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/View.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/View.java
deleted file mode 100644
index e2c4487..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/View.java
+++ /dev/null
@@ -1,470 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
-import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
-import com.google.cloud.dataflow.sdk.util.PCollectionViews;
-import com.google.cloud.dataflow.sdk.util.WindowedValue;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.cloud.dataflow.sdk.values.PCollectionView;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Transforms for creating {@link PCollectionView PCollectionViews} from
- * {@link PCollection PCollections} (to read them as side inputs).
- *
- * <p>While a {@link PCollection PCollection<ElemT>} has many values of type {@code ElemT} per
- * window, a {@link PCollectionView PCollectionView<ViewT>} has a single value of type
- * {@code ViewT} for each window. It can be thought of as a mapping from windows to values of
- * type {@code ViewT}. The transforms here represent ways of converting the {@code ElemT} values
- * in a window into a {@code ViewT} for that window.
- *
- * <p>When a {@link ParDo} tranform is processing a main input
- * element in a window {@code w} and a {@link PCollectionView} is read via
- * {@link DoFn.ProcessContext#sideInput}, the value of the view for {@code w} is
- * returned.
- *
- * <p>The SDK supports viewing a {@link PCollection}, per window, as a single value,
- * a {@link List}, an {@link Iterable}, a {@link Map}, or a multimap (iterable-valued {@link Map}).
- *
- * <p>For a {@link PCollection} that contains a single value of type {@code T}
- * per window, such as the output of {@link Combine#globally},
- * use {@link View#asSingleton()} to prepare it for use as a side input:
- *
- * <pre>
- * {@code
- * PCollectionView<T> output = someOtherPCollection
- * .apply(Combine.globally(...))
- * .apply(View.<T>asSingleton());
- * }
- * </pre>
- *
- * <p>For a small {@link PCollection} with windows that can fit entirely in memory,
- * use {@link View#asList()} to prepare it for use as a {@code List}.
- * When read as a side input, the entire list for a window will be cached in memory.
- *
- * <pre>
- * {@code
- * PCollectionView<List<T>> output =
- * smallPCollection.apply(View.<T>asList());
- * }
- * </pre>
- *
- * <p>If a {@link PCollection} of {@code KV<K, V>} is known to
- * have a single value per window for each key, then use {@link View#asMap()}
- * to view it as a {@code Map<K, V>}:
- *
- * <pre>
- * {@code
- * PCollectionView<Map<K, V> output =
- * somePCollection.apply(View.<K, V>asMap());
- * }
- * </pre>
- *
- * <p>Otherwise, to access a {@link PCollection} of {@code KV<K, V>} as a
- * {@code Map<K, Iterable<V>>} side input, use {@link View#asMultimap()}:
- *
- * <pre>
- * {@code
- * PCollectionView<Map<K, Iterable<V>> output =
- * somePCollection.apply(View.<K, Iterable<V>>asMap());
- * }
- * </pre>
- *
- * <p>To iterate over an entire window of a {@link PCollection} via
- * side input, use {@link View#asIterable()}:
- *
- * <pre>
- * {@code
- * PCollectionView<Iterable<T>> output =
- * somePCollection.apply(View.<T>asIterable());
- * }
- * </pre>
- *
- *
- * <p>Both {@link View#asMultimap()} and {@link View#asMap()} are useful
- * for implementing lookup based "joins" with the main input, when the
- * side input is small enough to fit into memory.
- *
- * <p>For example, if you represent a page on a website via some {@code Page} object and
- * have some type {@code UrlVisits} logging that a URL was visited, you could convert these
- * to more fully structured {@code PageVisit} objects using a side input, something like the
- * following:
- *
- * <pre>
- * {@code
- * PCollection<Page> pages = ... // pages fit into memory
- * PCollection<UrlVisit> urlVisits = ... // very large collection
- * final PCollectionView<Map<URL, Page>> = urlToPage
- * .apply(WithKeys.of( ... )) // extract the URL from the page
- * .apply(View.<URL, Page>asMap());
- *
- * PCollection PageVisits = urlVisits
- * .apply(ParDo.withSideInputs(urlToPage)
- * .of(new DoFn<UrlVisit, PageVisit>() {
- * {@literal @}Override
- * void processElement(ProcessContext context) {
- * UrlVisit urlVisit = context.element();
- * Page page = urlToPage.get(urlVisit.getUrl());
- * c.output(new PageVisit(page, urlVisit.getVisitData()));
- * }
- * }));
- * }
- * </pre>
- *
- * <p>See {@link ParDo#withSideInputs} for details on how to access
- * this variable inside a {@link ParDo} over another {@link PCollection}.
- */
-public class View {
-
- // Do not instantiate
- private View() { }
-
- /**
- * Returns a {@link AsSingleton} transform that takes a
- * {@link PCollection} with a single value per window
- * as input and produces a {@link PCollectionView} that returns
- * the value in the main input window when read as a side input.
- *
- * <pre>
- * {@code
- * PCollection<InputT> input = ...
- * CombineFn<InputT, OutputT> yourCombineFn = ...
- * PCollectionView<OutputT> output = input
- * .apply(Combine.globally(yourCombineFn))
- * .apply(View.<OutputT>asSingleton());
- * }</pre>
- *
- * <p>If the input {@link PCollection} is empty,
- * throws {@link java.util.NoSuchElementException} in the consuming
- * {@link DoFn}.
- *
- * <p>If the input {@link PCollection} contains more than one
- * element, throws {@link IllegalArgumentException} in the
- * consuming {@link DoFn}.
- */
- public static <T> AsSingleton<T> asSingleton() {
- return new AsSingleton<>();
- }
-
- /**
- * Returns a {@link View.AsList} transform that takes a {@link PCollection} and returns a
- * {@link PCollectionView} mapping each window to a {@link List} containing
- * all of the elements in the window.
- *
- * <p>The resulting list is required to fit in memory.
- */
- public static <T> AsList<T> asList() {
- return new AsList<>();
- }
-
- /**
- * Returns a {@link View.AsIterable} transform that takes a {@link PCollection} as input
- * and produces a {@link PCollectionView} mapping each window to an
- * {@link Iterable} of the values in that window.
- *
- * <p>The values of the {@link Iterable} for a window are not required to fit in memory,
- * but they may also not be effectively cached. If it is known that every window fits in memory,
- * and stronger caching is desired, use {@link #asList}.
- */
- public static <T> AsIterable<T> asIterable() {
- return new AsIterable<>();
- }
-
- /**
- * Returns a {@link View.AsMap} transform that takes a
- * {@link PCollection PCollection<KV<K V>>} as
- * input and produces a {@link PCollectionView} mapping each window to
- * a {@link Map Map>K, V>}. It is required that each key of the input be
- * associated with a single value, per window. If this is not the case, precede this
- * view with {@code Combine.perKey}, as in the example below, or alternatively
- * use {@link View#asMultimap()}.
- *
- * <pre>
- * {@code
- * PCollection<KV<K, V>> input = ...
- * CombineFn<V, OutputT> yourCombineFn = ...
- * PCollectionView<Map<K, OutputT>> output = input
- * .apply(Combine.perKey(yourCombineFn.<K>asKeyedFn()))
- * .apply(View.<K, OutputT>asMap());
- * }</pre>
- *
- * <p>Currently, the resulting map is required to fit into memory.
- */
- public static <K, V> AsMap<K, V> asMap() {
- return new AsMap<K, V>();
- }
-
- /**
- * Returns a {@link View.AsMultimap} transform that takes a
- * {@link PCollection PCollection<KV<K, V>>}
- * as input and produces a {@link PCollectionView} mapping
- * each window to its contents as a {@link Map Map<K, Iterable<V>>}
- * for use as a side input.
- * In contrast to {@link View#asMap()}, it is not required that the keys in the
- * input collection be unique.
- *
- * <pre>
- * {@code
- * PCollection<KV<K, V>> input = ... // maybe more than one occurrence of a some keys
- * PCollectionView<Map<K, V>> output = input.apply(View.<K, V>asMultimap());
- * }</pre>
- *
- * <p>Currently, the resulting map is required to fit into memory.
- */
- public static <K, V> AsMultimap<K, V> asMultimap() {
- return new AsMultimap<K, V>();
- }
-
- /**
- * Not intended for direct use by pipeline authors; public only so a {@link PipelineRunner} may
- * override its behavior.
- *
- * <p>See {@link View#asList()}.
- */
- public static class AsList<T> extends PTransform<PCollection<T>, PCollectionView<List<T>>> {
- private AsList() { }
-
- @Override
- public void validate(PCollection<T> input) {
- try {
- GroupByKey.applicableTo(input);
- } catch (IllegalStateException e) {
- throw new IllegalStateException("Unable to create a side-input view from input", e);
- }
- }
-
- @Override
- public PCollectionView<List<T>> apply(PCollection<T> input) {
- return input.apply(CreatePCollectionView.<T, List<T>>of(PCollectionViews.listView(
- input.getPipeline(), input.getWindowingStrategy(), input.getCoder())));
- }
- }
-
- /**
- * Not intended for direct use by pipeline authors; public only so a {@link PipelineRunner} may
- * override its behavior.
- *
- * <p>See {@link View#asIterable()}.
- */
- public static class AsIterable<T>
- extends PTransform<PCollection<T>, PCollectionView<Iterable<T>>> {
- private AsIterable() { }
-
- @Override
- public void validate(PCollection<T> input) {
- try {
- GroupByKey.applicableTo(input);
- } catch (IllegalStateException e) {
- throw new IllegalStateException("Unable to create a side-input view from input", e);
- }
- }
-
- @Override
- public PCollectionView<Iterable<T>> apply(PCollection<T> input) {
- return input.apply(CreatePCollectionView.<T, Iterable<T>>of(PCollectionViews.iterableView(
- input.getPipeline(), input.getWindowingStrategy(), input.getCoder())));
- }
- }
-
- /**
- * Not intended for direct use by pipeline authors; public only so a {@link PipelineRunner} may
- * override its behavior.
- *
- * <p>See {@link View#asSingleton()}.
- */
- public static class AsSingleton<T> extends PTransform<PCollection<T>, PCollectionView<T>> {
- private final T defaultValue;
- private final boolean hasDefault;
-
- private AsSingleton() {
- this.defaultValue = null;
- this.hasDefault = false;
- }
-
- private AsSingleton(T defaultValue) {
- this.defaultValue = defaultValue;
- this.hasDefault = true;
- }
-
- /**
- * Returns whether this transform has a default value.
- */
- public boolean hasDefaultValue() {
- return hasDefault;
- }
-
- /**
- * Returns the default value of this transform, or null if there isn't one.
- */
- public T defaultValue() {
- return defaultValue;
- }
-
- /**
- * Default value to return for windows with no value in them.
- */
- public AsSingleton<T> withDefaultValue(T defaultValue) {
- return new AsSingleton<>(defaultValue);
- }
-
- @Override
- public void validate(PCollection<T> input) {
- try {
- GroupByKey.applicableTo(input);
- } catch (IllegalStateException e) {
- throw new IllegalStateException("Unable to create a side-input view from input", e);
- }
- }
-
- @Override
- public PCollectionView<T> apply(PCollection<T> input) {
- return input.apply(CreatePCollectionView.<T, T>of(PCollectionViews.singletonView(
- input.getPipeline(),
- input.getWindowingStrategy(),
- hasDefault,
- defaultValue,
- input.getCoder())));
- }
- }
-
- /**
- * Not intended for direct use by pipeline authors; public only so a {@link PipelineRunner} may
- * override its behavior.
- *
- * <p>See {@link View#asMultimap()}.
- */
- public static class AsMultimap<K, V>
- extends PTransform<PCollection<KV<K, V>>, PCollectionView<Map<K, Iterable<V>>>> {
- private AsMultimap() { }
-
- @Override
- public void validate(PCollection<KV<K, V>> input) {
- try {
- GroupByKey.applicableTo(input);
- } catch (IllegalStateException e) {
- throw new IllegalStateException("Unable to create a side-input view from input", e);
- }
- }
-
- @Override
- public PCollectionView<Map<K, Iterable<V>>> apply(PCollection<KV<K, V>> input) {
- return input.apply(CreatePCollectionView.<KV<K, V>, Map<K, Iterable<V>>>of(
- PCollectionViews.multimapView(
- input.getPipeline(),
- input.getWindowingStrategy(),
- input.getCoder())));
- }
- }
-
- /**
- * Not intended for direct use by pipeline authors; public only so a {@link PipelineRunner} may
- * override its behavior.
- *
- * <p>See {@link View#asMap()}.
- */
- public static class AsMap<K, V>
- extends PTransform<PCollection<KV<K, V>>, PCollectionView<Map<K, V>>> {
- private AsMap() { }
-
- /**
- * @deprecated this method simply returns this AsMap unmodified
- */
- @Deprecated()
- public AsMap<K, V> withSingletonValues() {
- return this;
- }
-
- @Override
- public void validate(PCollection<KV<K, V>> input) {
- try {
- GroupByKey.applicableTo(input);
- } catch (IllegalStateException e) {
- throw new IllegalStateException("Unable to create a side-input view from input", e);
- }
- }
-
- @Override
- public PCollectionView<Map<K, V>> apply(PCollection<KV<K, V>> input) {
- return input.apply(CreatePCollectionView.<KV<K, V>, Map<K, V>>of(
- PCollectionViews.mapView(
- input.getPipeline(),
- input.getWindowingStrategy(),
- input.getCoder())));
- }
- }
-
- ////////////////////////////////////////////////////////////////////////////
- // Internal details below
-
- /**
- * Creates a primitive {@link PCollectionView}.
- *
- * <p>For internal use only by runner implementors.
- *
- * @param <ElemT> The type of the elements of the input PCollection
- * @param <ViewT> The type associated with the {@link PCollectionView} used as a side input
- */
- public static class CreatePCollectionView<ElemT, ViewT>
- extends PTransform<PCollection<ElemT>, PCollectionView<ViewT>> {
- private PCollectionView<ViewT> view;
-
- private CreatePCollectionView(PCollectionView<ViewT> view) {
- this.view = view;
- }
-
- public static <ElemT, ViewT> CreatePCollectionView<ElemT, ViewT> of(
- PCollectionView<ViewT> view) {
- return new CreatePCollectionView<>(view);
- }
-
- public PCollectionView<ViewT> getView() {
- return view;
- }
-
- @Override
- public PCollectionView<ViewT> apply(PCollection<ElemT> input) {
- return view;
- }
-
- static {
- DirectPipelineRunner.registerDefaultTransformEvaluator(
- CreatePCollectionView.class,
- new DirectPipelineRunner.TransformEvaluator<CreatePCollectionView>() {
- @SuppressWarnings("rawtypes")
- @Override
- public void evaluate(
- CreatePCollectionView transform,
- DirectPipelineRunner.EvaluationContext context) {
- evaluateTyped(transform, context);
- }
-
- private <ElemT, ViewT> void evaluateTyped(
- CreatePCollectionView<ElemT, ViewT> transform,
- DirectPipelineRunner.EvaluationContext context) {
- List<WindowedValue<ElemT>> elems =
- context.getPCollectionWindowedValues(context.getInput(transform));
- context.setPCollectionView(context.getOutput(transform), elems);
- }
- });
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithKeys.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithKeys.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithKeys.java
deleted file mode 100644
index c06795c..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithKeys.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import com.google.cloud.dataflow.sdk.coders.CannotProvideCoderException;
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.CoderRegistry;
-import com.google.cloud.dataflow.sdk.coders.KvCoder;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
-
-/**
- * {@code WithKeys<K, V>} takes a {@code PCollection<V>}, and either a
- * constant key of type {@code K} or a function from {@code V} to
- * {@code K}, and returns a {@code PCollection<KV<K, V>>}, where each
- * of the values in the input {@code PCollection} has been paired with
- * either the constant key or a key computed from the value.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<String> words = ...;
- * PCollection<KV<Integer, String>> lengthsToWords =
- * words.apply(WithKeys.of(new SerializableFunction<String, Integer>() {
- * public Integer apply(String s) { return s.length(); } }));
- * } </pre>
- *
- * <p>Each output element has the same timestamp and is in the same windows
- * as its corresponding input element, and the output {@code PCollection}
- * has the same
- * {@link com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn}
- * associated with it as the input.
- *
- * @param <K> the type of the keys in the output {@code PCollection}
- * @param <V> the type of the elements in the input
- * {@code PCollection} and the values in the output
- * {@code PCollection}
- */
-public class WithKeys<K, V> extends PTransform<PCollection<V>,
- PCollection<KV<K, V>>> {
- /**
- * Returns a {@code PTransform} that takes a {@code PCollection<V>}
- * and returns a {@code PCollection<KV<K, V>>}, where each of the
- * values in the input {@code PCollection} has been paired with a
- * key computed from the value by invoking the given
- * {@code SerializableFunction}.
- *
- * <p>If using a lambda in Java 8, {@link #withKeyType(TypeDescriptor)} must
- * be called on the result {@link PTransform}.
- */
- public static <K, V> WithKeys<K, V> of(SerializableFunction<V, K> fn) {
- return new WithKeys<>(fn, null);
- }
-
- /**
- * Returns a {@code PTransform} that takes a {@code PCollection<V>}
- * and returns a {@code PCollection<KV<K, V>>}, where each of the
- * values in the input {@code PCollection} has been paired with the
- * given key.
- */
- @SuppressWarnings("unchecked")
- public static <K, V> WithKeys<K, V> of(final K key) {
- return new WithKeys<>(
- new SerializableFunction<V, K>() {
- @Override
- public K apply(V value) {
- return key;
- }
- },
- (Class<K>) (key == null ? null : key.getClass()));
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- private SerializableFunction<V, K> fn;
- private transient Class<K> keyClass;
-
- private WithKeys(SerializableFunction<V, K> fn, Class<K> keyClass) {
- this.fn = fn;
- this.keyClass = keyClass;
- }
-
- /**
- * Return a {@link WithKeys} that is like this one with the specified key type descriptor.
- *
- * For use with lambdas in Java 8, either this method must be called with an appropriate type
- * descriptor or {@link PCollection#setCoder(Coder)} must be called on the output
- * {@link PCollection}.
- */
- public WithKeys<K, V> withKeyType(TypeDescriptor<K> keyType) {
- // Safe cast
- @SuppressWarnings("unchecked")
- Class<K> rawType = (Class<K>) keyType.getRawType();
- return new WithKeys<>(fn, rawType);
- }
-
- @Override
- public PCollection<KV<K, V>> apply(PCollection<V> in) {
- PCollection<KV<K, V>> result =
- in.apply(ParDo.named("AddKeys")
- .of(new DoFn<V, KV<K, V>>() {
- @Override
- public void processElement(ProcessContext c) {
- c.output(KV.of(fn.apply(c.element()),
- c.element()));
- }
- }));
-
- try {
- Coder<K> keyCoder;
- CoderRegistry coderRegistry = in.getPipeline().getCoderRegistry();
- if (keyClass == null) {
- keyCoder = coderRegistry.getDefaultOutputCoder(fn, in.getCoder());
- } else {
- keyCoder = coderRegistry.getDefaultCoder(TypeDescriptor.of(keyClass));
- }
- // TODO: Remove when we can set the coder inference context.
- result.setCoder(KvCoder.of(keyCoder, in.getCoder()));
- } catch (CannotProvideCoderException exc) {
- // let lazy coder inference have a try
- }
-
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithTimestamps.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithTimestamps.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithTimestamps.java
deleted file mode 100644
index 85a93bf..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/WithTimestamps.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.cloud.dataflow.sdk.io.Source;
-import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
-import com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-
-import org.joda.time.Duration;
-import org.joda.time.Instant;
-
-/**
- * A {@link PTransform} for assigning timestamps to all the elements of a {@link PCollection}.
- *
- * <p>Timestamps are used to assign {@link BoundedWindow Windows} to elements within the
- * {@link Window#into(com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn)}
- * {@link PTransform}. Assigning timestamps is useful when the input data set comes from a
- * {@link Source} without implicit timestamps (such as
- * {@link com.google.cloud.dataflow.sdk.io.TextIO.Read TextIO}).
- *
- */
-public class WithTimestamps<T> extends PTransform<PCollection<T>, PCollection<T>> {
- /**
- * For a {@link SerializableFunction} {@code fn} from {@code T} to {@link Instant}, outputs a
- * {@link PTransform} that takes an input {@link PCollection PCollection<T>} and outputs a
- * {@link PCollection PCollection<T>} containing every element {@code v} in the input where
- * each element is output with a timestamp obtained as the result of {@code fn.apply(v)}.
- *
- * <p>If the input {@link PCollection} elements have timestamps, the output timestamp for each
- * element must not be before the input element's timestamp minus the value of
- * {@link #getAllowedTimestampSkew()}. If an output timestamp is before this time, the transform
- * will throw an {@link IllegalArgumentException} when executed. Use
- * {@link #withAllowedTimestampSkew(Duration)} to update the allowed skew.
- *
- * <p>Each output element will be in the same windows as the input element. If a new window based
- * on the new output timestamp is desired, apply a new instance of {@link Window#into(WindowFn)}.
- *
- * <p>This transform will fail at execution time with a {@link NullPointerException} if for any
- * input element the result of {@code fn.apply(v)} is {@code null}.
- *
- * <p>Example of use in Java 8:
- * <pre>{@code
- * PCollection<Record> timestampedRecords = records.apply(
- * WithTimestamps.of((Record rec) -> rec.getInstant());
- * }</pre>
- */
- public static <T> WithTimestamps<T> of(SerializableFunction<T, Instant> fn) {
- return new WithTimestamps<>(fn, Duration.ZERO);
- }
-
- ///////////////////////////////////////////////////////////////////
-
- private final SerializableFunction<T, Instant> fn;
- private final Duration allowedTimestampSkew;
-
- private WithTimestamps(SerializableFunction<T, Instant> fn, Duration allowedTimestampSkew) {
- this.fn = checkNotNull(fn, "WithTimestamps fn cannot be null");
- this.allowedTimestampSkew = allowedTimestampSkew;
- }
-
- /**
- * Return a new WithTimestamps like this one with updated allowed timestamp skew, which is the
- * maximum duration that timestamps can be shifted backward. Does not modify this object.
- *
- * <p>The default value is {@code Duration.ZERO}, allowing timestamps to only be shifted into the
- * future. For infinite skew, use {@code new Duration(Long.MAX_VALUE)}.
- */
- public WithTimestamps<T> withAllowedTimestampSkew(Duration allowedTimestampSkew) {
- return new WithTimestamps<>(this.fn, allowedTimestampSkew);
- }
-
- /**
- * Returns the allowed timestamp skew duration, which is the maximum
- * duration that timestamps can be shifted backwards from the timestamp of the input element.
- *
- * @see DoFn#getAllowedTimestampSkew()
- */
- public Duration getAllowedTimestampSkew() {
- return allowedTimestampSkew;
- }
-
- @Override
- public PCollection<T> apply(PCollection<T> input) {
- return input
- .apply(ParDo.named("AddTimestamps").of(new AddTimestampsDoFn<T>(fn, allowedTimestampSkew)))
- .setTypeDescriptorInternal(input.getTypeDescriptor());
- }
-
- private static class AddTimestampsDoFn<T> extends DoFn<T, T> {
- private final SerializableFunction<T, Instant> fn;
- private final Duration allowedTimestampSkew;
-
- public AddTimestampsDoFn(SerializableFunction<T, Instant> fn, Duration allowedTimestampSkew) {
- this.fn = fn;
- this.allowedTimestampSkew = allowedTimestampSkew;
- }
-
- @Override
- public void processElement(ProcessContext c) {
- Instant timestamp = fn.apply(c.element());
- checkNotNull(
- timestamp, "Timestamps for WithTimestamps cannot be null. Timestamp provided by %s.", fn);
- c.outputWithTimestamp(c.element(), timestamp);
- }
-
- @Override
- public Duration getAllowedTimestampSkew() {
- return allowedTimestampSkew;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Write.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Write.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Write.java
deleted file mode 100644
index 5cf655a..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Write.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
- * in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms;
-
-/**
- * A backwards-compatible {@code Write} class that simply inherits from the
- * {@link com.google.cloud.dataflow.sdk.io.Write} class that should be used instead.
- *
- * @deprecated: use {@link com.google.cloud.dataflow.sdk.io.Write} from the
- * {@code com.google.cloud.dataflow.sdk.io} package instead.
- */
-@Deprecated
-public class Write extends com.google.cloud.dataflow.sdk.io.Write {
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/DisplayData.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/DisplayData.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/DisplayData.java
deleted file mode 100644
index dadc730..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/DisplayData.java
+++ /dev/null
@@ -1,530 +0,0 @@
-/*
- * Copyright (C) 2016 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms.display;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.cloud.dataflow.sdk.transforms.DoFn;
-import com.google.cloud.dataflow.sdk.transforms.PTransform;
-import com.google.cloud.dataflow.sdk.transforms.ParDo;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
-import com.fasterxml.jackson.annotation.JsonGetter;
-import com.fasterxml.jackson.annotation.JsonInclude;
-
-import org.apache.avro.reflect.Nullable;
-import org.joda.time.Duration;
-import org.joda.time.Instant;
-import org.joda.time.format.DateTimeFormatter;
-import org.joda.time.format.ISODateTimeFormat;
-
-import java.util.Collection;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * Static display metadata associated with a pipeline component. Display data is useful for
- * pipeline runner UIs and diagnostic dashboards to display details about
- * {@link PTransform PTransforms} that make up a pipeline.
- *
- * <p>Components specify their display data by implementing the {@link HasDisplayData}
- * interface.
- */
-public class DisplayData {
- private static final DisplayData EMPTY = new DisplayData(Maps.<Identifier, Item>newHashMap());
- private static final DateTimeFormatter TIMESTAMP_FORMATTER = ISODateTimeFormat.dateTime();
-
- private final ImmutableMap<Identifier, Item> entries;
-
- private DisplayData(Map<Identifier, Item> entries) {
- this.entries = ImmutableMap.copyOf(entries);
- }
-
- /**
- * Default empty {@link DisplayData} instance.
- */
- public static DisplayData none() {
- return EMPTY;
- }
-
- /**
- * Collect the {@link DisplayData} from a component. This will traverse all subcomponents
- * specified via {@link Builder#include} in the given component. Data in this component will be in
- * a namespace derived from the component.
- */
- public static DisplayData from(HasDisplayData component) {
- checkNotNull(component);
- return InternalBuilder.forRoot(component).build();
- }
-
- public Collection<Item> items() {
- return entries.values();
- }
-
- public Map<Identifier, Item> asMap() {
- return entries;
- }
-
- @Override
- public String toString() {
- StringBuilder builder = new StringBuilder();
- boolean isFirstLine = true;
- for (Map.Entry<Identifier, Item> entry : entries.entrySet()) {
- if (isFirstLine) {
- isFirstLine = false;
- } else {
- builder.append("\n");
- }
-
- builder.append(entry);
- }
-
- return builder.toString();
- }
-
- /**
- * Utility to build up display metadata from a component and its included
- * subcomponents.
- */
- public interface Builder {
- /**
- * Include display metadata from the specified subcomponent. For example, a {@link ParDo}
- * transform includes display metadata from the encapsulated {@link DoFn}.
- *
- * @return A builder instance to continue to build in a fluent-style.
- */
- Builder include(HasDisplayData subComponent);
-
- /**
- * Register the given string display metadata. The metadata item will be registered with type
- * {@link DisplayData.Type#STRING}, and is identified by the specified key and namespace from
- * the current transform or component.
- */
- ItemBuilder add(String key, String value);
-
- /**
- * Register the given numeric display metadata. The metadata item will be registered with type
- * {@link DisplayData.Type#INTEGER}, and is identified by the specified key and namespace from
- * the current transform or component.
- */
- ItemBuilder add(String key, long value);
-
- /**
- * Register the given floating point display metadata. The metadata item will be registered with
- * type {@link DisplayData.Type#FLOAT}, and is identified by the specified key and namespace
- * from the current transform or component.
- */
- ItemBuilder add(String key, double value);
-
- /**
- * Register the given timestamp display metadata. The metadata item will be registered with type
- * {@link DisplayData.Type#TIMESTAMP}, and is identified by the specified key and namespace from
- * the current transform or component.
- */
- ItemBuilder add(String key, Instant value);
-
- /**
- * Register the given duration display metadata. The metadata item will be registered with type
- * {@link DisplayData.Type#DURATION}, and is identified by the specified key and namespace from
- * the current transform or component.
- */
- ItemBuilder add(String key, Duration value);
-
- /**
- * Register the given class display metadata. The metadata item will be registered with type
- * {@link DisplayData.Type#JAVA_CLASS}, and is identified by the specified key and namespace
- * from the current transform or component.
- */
- ItemBuilder add(String key, Class<?> value);
- }
-
- /**
- * Utility to append optional fields to display metadata, or register additional display metadata
- * items.
- */
- public interface ItemBuilder extends Builder {
- /**
- * Add a human-readable label to describe the most-recently added metadata field.
- * A label is optional; if unspecified, UIs should display the metadata key to identify the
- * display item.
- *
- * <p>Specifying a null value will clear the label if it was previously defined.
- */
- ItemBuilder withLabel(@Nullable String label);
-
- /**
- * Add a link URL to the most-recently added display metadata. A link URL is optional and
- * can be provided to point the reader to additional details about the metadata.
- *
- * <p>Specifying a null value will clear the URL if it was previously defined.
- */
- ItemBuilder withLinkUrl(@Nullable String url);
- }
-
- /**
- * A display metadata item. DisplayData items are registered via {@link Builder#add} within
- * {@link HasDisplayData#populateDisplayData} implementations. Each metadata item is uniquely
- * identified by the specified key and namespace generated from the registering component's
- * class name.
- */
- public static class Item {
- private final String key;
- private final String ns;
- private final Type type;
- private final String value;
- private final String shortValue;
- private final String label;
- private final String url;
-
- private static <T> Item create(String namespace, String key, Type type, T value) {
- FormattedItemValue formatted = type.format(value);
- return new Item(
- namespace, key, type, formatted.getLongValue(), formatted.getShortValue(), null, null);
- }
-
- private Item(
- String namespace,
- String key,
- Type type,
- String value,
- String shortValue,
- String url,
- String label) {
- this.ns = namespace;
- this.key = key;
- this.type = type;
- this.value = value;
- this.shortValue = shortValue;
- this.url = url;
- this.label = label;
- }
-
- @JsonGetter("namespace")
- public String getNamespace() {
- return ns;
- }
-
- @JsonGetter("key")
- public String getKey() {
- return key;
- }
-
- /**
- * Retrieve the {@link DisplayData.Type} of display metadata. All metadata conforms to a
- * predefined set of allowed types.
- */
- @JsonGetter("type")
- public Type getType() {
- return type;
- }
-
- /**
- * Retrieve the value of the metadata item.
- */
- @JsonGetter("value")
- public String getValue() {
- return value;
- }
-
- /**
- * Return the optional short value for an item. Types may provide a short-value to displayed
- * instead of or in addition to the full {@link Item#value}.
- *
- * <p>Some display data types will not provide a short value, in which case the return value
- * will be null.
- */
- @JsonGetter("shortValue")
- @JsonInclude(JsonInclude.Include.NON_NULL)
- @Nullable
- public String getShortValue() {
- return shortValue;
- }
-
- /**
- * Retrieve the optional label for an item. The label is a human-readable description of what
- * the metadata represents. UIs may choose to display the label instead of the item key.
- *
- * <p>If no label was specified, this will return {@code null}.
- */
- @JsonGetter("label")
- @JsonInclude(JsonInclude.Include.NON_NULL)
- @Nullable
- public String getLabel() {
- return label;
- }
-
- /**
- * Retrieve the optional link URL for an item. The URL points to an address where the reader
- * can find additional context for the display metadata.
- *
- * <p>If no URL was specified, this will return {@code null}.
- */
- @JsonGetter("linkUrl")
- @JsonInclude(JsonInclude.Include.NON_NULL)
- @Nullable
- public String getLinkUrl() {
- return url;
- }
-
- @Override
- public String toString() {
- return getValue();
- }
-
- private Item withLabel(String label) {
- return new Item(this.ns, this.key, this.type, this.value, this.shortValue, this.url, label);
- }
-
- private Item withUrl(String url) {
- return new Item(this.ns, this.key, this.type, this.value, this.shortValue, url, this.label);
- }
- }
-
- /**
- * Unique identifier for a display metadata item within a component.
- * Identifiers are composed of the key they are registered with and a namespace generated from
- * the class of the component which registered the item.
- *
- * <p>Display metadata registered with the same key from different components will have different
- * namespaces and thus will both be represented in the composed {@link DisplayData}. If a
- * single component registers multiple metadata items with the same key, only the most recent
- * item will be retained; previous versions are discarded.
- */
- public static class Identifier {
- private final String ns;
- private final String key;
-
- static Identifier of(Class<?> namespace, String key) {
- return new Identifier(namespace.getName(), key);
- }
-
- private Identifier(String ns, String key) {
- this.ns = ns;
- this.key = key;
- }
-
- public String getNamespace() {
- return ns;
- }
-
- public String getKey() {
- return key;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj instanceof Identifier) {
- Identifier that = (Identifier) obj;
- return Objects.equals(this.ns, that.ns)
- && Objects.equals(this.key, that.key);
- }
-
- return false;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(ns, key);
- }
-
- @Override
- public String toString() {
- return String.format("%s:%s", ns, key);
- }
- }
-
- /**
- * Display metadata type.
- */
- enum Type {
- STRING {
- @Override
- FormattedItemValue format(Object value) {
- return new FormattedItemValue((String) value);
- }
- },
- INTEGER {
- @Override
- FormattedItemValue format(Object value) {
- return new FormattedItemValue(Long.toString((long) value));
- }
- },
- FLOAT {
- @Override
- FormattedItemValue format(Object value) {
- return new FormattedItemValue(Double.toString((Double) value));
- }
- },
- TIMESTAMP() {
- @Override
- FormattedItemValue format(Object value) {
- return new FormattedItemValue((TIMESTAMP_FORMATTER.print((Instant) value)));
- }
- },
- DURATION {
- @Override
- FormattedItemValue format(Object value) {
- return new FormattedItemValue(Long.toString(((Duration) value).getMillis()));
- }
- },
- JAVA_CLASS {
- @Override
- FormattedItemValue format(Object value) {
- Class<?> clazz = (Class<?>) value;
- return new FormattedItemValue(clazz.getName(), clazz.getSimpleName());
- }
- };
-
- /**
- * Format the display metadata value into a long string representation, and optionally
- * a shorter representation for display.
- *
- * <p>Internal-only. Value objects can be safely cast to the expected Java type.
- */
- abstract FormattedItemValue format(Object value);
- }
-
- private static class FormattedItemValue {
- private final String shortValue;
- private final String longValue;
-
- private FormattedItemValue(String longValue) {
- this(longValue, null);
- }
-
- private FormattedItemValue(String longValue, String shortValue) {
- this.longValue = longValue;
- this.shortValue = shortValue;
- }
-
- private String getLongValue () {
- return this.longValue;
- }
-
- private String getShortValue() {
- return this.shortValue;
- }
- }
-
- private static class InternalBuilder implements ItemBuilder {
- private final Map<Identifier, Item> entries;
- private final Set<Object> visited;
-
- private Class<?> latestNs;
- private Item latestItem;
- private Identifier latestIdentifier;
-
- private InternalBuilder() {
- this.entries = Maps.newHashMap();
- this.visited = Sets.newIdentityHashSet();
- }
-
- private static InternalBuilder forRoot(HasDisplayData instance) {
- InternalBuilder builder = new InternalBuilder();
- builder.include(instance);
- return builder;
- }
-
- @Override
- public Builder include(HasDisplayData subComponent) {
- checkNotNull(subComponent);
- boolean newComponent = visited.add(subComponent);
- if (newComponent) {
- Class prevNs = this.latestNs;
- this.latestNs = subComponent.getClass();
- subComponent.populateDisplayData(this);
- this.latestNs = prevNs;
- }
-
- return this;
- }
-
- @Override
- public ItemBuilder add(String key, String value) {
- checkNotNull(value);
- return addItem(key, Type.STRING, value);
- }
-
- @Override
- public ItemBuilder add(String key, long value) {
- return addItem(key, Type.INTEGER, value);
- }
-
- @Override
- public ItemBuilder add(String key, double value) {
- return addItem(key, Type.FLOAT, value);
- }
-
- @Override
- public ItemBuilder add(String key, Instant value) {
- checkNotNull(value);
- return addItem(key, Type.TIMESTAMP, value);
- }
-
- @Override
- public ItemBuilder add(String key, Duration value) {
- checkNotNull(value);
- return addItem(key, Type.DURATION, value);
- }
-
- @Override
- public ItemBuilder add(String key, Class<?> value) {
- checkNotNull(value);
- return addItem(key, Type.JAVA_CLASS, value);
- }
-
- private <T> ItemBuilder addItem(String key, Type type, T value) {
- checkNotNull(key);
- checkArgument(!key.isEmpty());
-
- Identifier id = Identifier.of(latestNs, key);
- if (entries.containsKey(id)) {
- throw new IllegalArgumentException("DisplayData key already exists. All display data "
- + "for a component must be registered with a unique key.\nKey: " + id);
- }
- Item item = Item.create(id.getNamespace(), key, type, value);
- entries.put(id, item);
-
- latestItem = item;
- latestIdentifier = id;
-
- return this;
- }
-
- @Override
- public ItemBuilder withLabel(String label) {
- latestItem = latestItem.withLabel(label);
- entries.put(latestIdentifier, latestItem);
- return this;
- }
-
- @Override
- public ItemBuilder withLinkUrl(String url) {
- latestItem = latestItem.withUrl(url);
- entries.put(latestIdentifier, latestItem);
- return this;
- }
-
- private DisplayData build() {
- return new DisplayData(this.entries);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/HasDisplayData.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/HasDisplayData.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/HasDisplayData.java
deleted file mode 100644
index b2eca3d..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/display/HasDisplayData.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Copyright (C) 2016 Google Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package com.google.cloud.dataflow.sdk.transforms.display;
-
-import com.google.cloud.dataflow.sdk.transforms.PTransform;
-
-/**
- * Marker interface for {@link PTransform PTransforms} and components used within
- * {@link PTransform PTransforms} to specify display metadata to be used within UIs and diagnostic
- * tools.
- *
- * <p>Display metadata is optional and may be collected during pipeline construction. It should
- * only be used to informational purposes. Tools and components should not assume that display data
- * will always be collected, or that collected display data will always be displayed.
- */
-public interface HasDisplayData {
- /**
- * Register display metadata for the given transform or component. Metadata can be registered
- * directly on the provided builder, as well as via included sub-components.
- *
- * <pre>
- * {@code
- * @Override
- * public void populateDisplayData(DisplayData.Builder builder) {
- * builder
- * .include(subComponent)
- * .add("minFilter", 42)
- * .add("topic", "projects/myproject/topics/mytopic")
- * .withLabel("Pub/Sub Topic")
- * .add("serviceInstance", "myservice.com/fizzbang")
- * .withLinkUrl("http://www.myservice.com/fizzbang");
- * }
- * }
- * </pre>
- *
- * @param builder The builder to populate with display metadata.
- */
- void populateDisplayData(DisplayData.Builder builder);
-}