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:50 UTC
[26/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/ApproximateQuantiles.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateQuantiles.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateQuantiles.java
deleted file mode 100644
index 57dd510..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateQuantiles.java
+++ /dev/null
@@ -1,766 +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.BigEndianIntegerCoder;
-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.util.WeightedValue;
-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.Iterators;
-import com.google.common.collect.Lists;
-import com.google.common.collect.UnmodifiableIterator;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Iterator;
-import java.util.List;
-import java.util.PriorityQueue;
-
-import javax.annotation.Nullable;
-
-/**
- * {@code PTransform}s for getting an idea of a {@code PCollection}'s
- * data distribution using approximate {@code N}-tiles (e.g. quartiles,
- * percentiles, etc.), either globally or per-key.
- */
-public class ApproximateQuantiles {
- private ApproximateQuantiles() {
- // do not instantiate
- }
-
- /**
- * Returns a {@code PTransform} that takes a {@code PCollection<T>}
- * and returns a {@code PCollection<List<T>>} whose single value is a
- * {@code List} of the approximate {@code N}-tiles of the elements
- * of the input {@code PCollection}. This gives an idea of the
- * distribution of the input elements.
- *
- * <p>The computed {@code List} is of size {@code numQuantiles},
- * and contains the input elements' minimum value,
- * {@code numQuantiles-2} intermediate values, and maximum value, in
- * sorted order, using the given {@code Comparator} to order values.
- * To compute traditional {@code N}-tiles, one should use
- * {@code ApproximateQuantiles.globally(compareFn, N+1)}.
- *
- * <p>If there are fewer input elements than {@code numQuantiles},
- * then the result {@code List} will contain all the input elements,
- * in sorted order.
- *
- * <p>The argument {@code Comparator} must be {@code Serializable}.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<String> pc = ...;
- * PCollection<List<String>> quantiles =
- * pc.apply(ApproximateQuantiles.globally(stringCompareFn, 11));
- * } </pre>
- *
- * @param <T> the type of the elements in the input {@code PCollection}
- * @param numQuantiles the number of elements in the resulting
- * quantile values {@code List}
- * @param compareFn the function to use to order the elements
- */
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- PTransform<PCollection<T>, PCollection<List<T>>> globally(
- int numQuantiles, ComparatorT compareFn) {
- return Combine.globally(
- ApproximateQuantilesCombineFn.create(numQuantiles, compareFn));
- }
-
- /**
- * Like {@link #globally(int, Comparator)}, but sorts using the
- * elements' natural ordering.
- *
- * @param <T> the type of the elements in the input {@code PCollection}
- * @param numQuantiles the number of elements in the resulting
- * quantile values {@code List}
- */
- public static <T extends Comparable<T>>
- PTransform<PCollection<T>, PCollection<List<T>>> globally(int numQuantiles) {
- return Combine.globally(
- ApproximateQuantilesCombineFn.<T>create(numQuantiles));
- }
-
- /**
- * Returns a {@code PTransform} that takes a
- * {@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 a {@code List} of the approximate
- * {@code N}-tiles of the values associated with that key in the
- * input {@code PCollection}. This gives an idea of the
- * distribution of the input values for each key.
- *
- * <p>Each of the computed {@code List}s is of size {@code numQuantiles},
- * and contains the input values' minimum value,
- * {@code numQuantiles-2} intermediate values, and maximum value, in
- * sorted order, using the given {@code Comparator} to order values.
- * To compute traditional {@code N}-tiles, one should use
- * {@code ApproximateQuantiles.perKey(compareFn, N+1)}.
- *
- * <p>If a key has fewer than {@code numQuantiles} values
- * associated with it, then that key's output {@code List} will
- * contain all the key's input values, in sorted order.
- *
- * <p>The argument {@code Comparator} must be {@code Serializable}.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<Integer, String>> pc = ...;
- * PCollection<KV<Integer, List<String>>> quantilesPerKey =
- * pc.apply(ApproximateQuantiles.<Integer, String>perKey(stringCompareFn, 11));
- * } </pre>
- *
- * <p>See {@link Combine.PerKey} for how this affects timestamps and windowing.
- *
- * @param <K> the type of the keys in the input and output
- * {@code PCollection}s
- * @param <V> the type of the values in the input {@code PCollection}
- * @param numQuantiles the number of elements in the resulting
- * quantile values {@code List}
- * @param compareFn the function to use to order the elements
- */
- public static <K, V, ComparatorT extends Comparator<V> & Serializable>
- PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>>
- perKey(int numQuantiles, ComparatorT compareFn) {
- return Combine.perKey(
- ApproximateQuantilesCombineFn.create(numQuantiles, compareFn)
- .<K>asKeyedFn());
- }
-
- /**
- * Like {@link #perKey(int, Comparator)}, but sorts
- * values using the their natural ordering.
- *
- * @param <K> the type of the keys in the input and output
- * {@code PCollection}s
- * @param <V> the type of the values in the input {@code PCollection}
- * @param numQuantiles the number of elements in the resulting
- * quantile values {@code List}
- */
- public static <K, V extends Comparable<V>>
- PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>>
- perKey(int numQuantiles) {
- return Combine.perKey(
- ApproximateQuantilesCombineFn.<V>create(numQuantiles)
- .<K>asKeyedFn());
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * The {@code ApproximateQuantilesCombineFn} combiner gives an idea
- * of the distribution of a collection of values using approximate
- * {@code N}-tiles. The output of this combiner is a {@code List}
- * of size {@code numQuantiles}, containing the input values'
- * minimum value, {@code numQuantiles-2} intermediate values, and
- * maximum value, in sorted order, so for traditional
- * {@code N}-tiles, one should use
- * {@code ApproximateQuantilesCombineFn#create(N+1)}.
- *
- * <p>If there are fewer values to combine than
- * {@code numQuantiles}, then the result {@code List} will contain all the
- * values being combined, in sorted order.
- *
- * <p>Values are ordered using either a specified
- * {@code Comparator} or the values' natural ordering.
- *
- * <p>To evaluate the quantiles we use the "New Algorithm" described here:
- * <pre>
- * [MRL98] Manku, Rajagopalan & Lindsay, "Approximate Medians and other
- * Quantiles in One Pass and with Limited Memory", Proc. 1998 ACM
- * SIGMOD, Vol 27, No 2, p 426-435, June 1998.
- * http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.6.6513&rep=rep1&type=pdf
- * </pre>
- *
- * <p>The default error bound is {@code 1 / N}, though in practice
- * the accuracy tends to be much better. <p>See
- * {@link #create(int, Comparator, long, double)} for
- * more information about the meaning of {@code epsilon}, and
- * {@link #withEpsilon} for a convenient way to adjust it.
- *
- * @param <T> the type of the values being combined
- */
- public static class ApproximateQuantilesCombineFn
- <T, ComparatorT extends Comparator<T> & Serializable>
- extends AccumulatingCombineFn<T, QuantileState<T, ComparatorT>, List<T>> {
-
- /**
- * The cost (in time and space) to compute quantiles to a given
- * accuracy is a function of the total number of elements in the
- * data set. If an estimate is not known or specified, we use
- * this as an upper bound. If this is too low, errors may exceed
- * the requested tolerance; if too high, efficiency may be
- * non-optimal. The impact is logarithmic with respect to this
- * value, so this default should be fine for most uses.
- */
- public static final long DEFAULT_MAX_NUM_ELEMENTS = (long) 1e9;
-
- /** The comparison function to use. */
- private final ComparatorT compareFn;
-
- /**
- * Number of quantiles to produce. The size of the final output
- * list, including the minimum and maximum, is numQuantiles.
- */
- private final int numQuantiles;
-
- /** The size of the buffers, corresponding to k in the referenced paper. */
- private final int bufferSize;
-
- /** The number of buffers, corresponding to b in the referenced paper. */
- private final int numBuffers;
-
- private final long maxNumElements;
-
- private ApproximateQuantilesCombineFn(
- int numQuantiles,
- ComparatorT compareFn,
- int bufferSize,
- int numBuffers,
- long maxNumElements) {
- Preconditions.checkArgument(numQuantiles >= 2);
- Preconditions.checkArgument(bufferSize >= 2);
- Preconditions.checkArgument(numBuffers >= 2);
- this.numQuantiles = numQuantiles;
- this.compareFn = compareFn;
- this.bufferSize = bufferSize;
- this.numBuffers = numBuffers;
- this.maxNumElements = maxNumElements;
- }
-
- /**
- * Returns an approximate quantiles combiner with the given
- * {@code compareFn} and desired number of quantiles. A total of
- * {@code numQuantiles} elements will appear in the output list,
- * including the minimum and maximum.
- *
- * <p>The {@code Comparator} must be {@code Serializable}.
- *
- * <p>The default error bound is {@code 1 / numQuantiles}, which
- * holds as long as the number of elements is less than
- * {@link #DEFAULT_MAX_NUM_ELEMENTS}.
- */
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- ApproximateQuantilesCombineFn<T, ComparatorT> create(
- int numQuantiles, ComparatorT compareFn) {
- return create(
- numQuantiles, compareFn, DEFAULT_MAX_NUM_ELEMENTS, 1.0 / numQuantiles);
- }
-
- /**
- * Like {@link #create(int, Comparator)}, but sorts values using their natural ordering.
- */
- public static <T extends Comparable<T>>
- ApproximateQuantilesCombineFn<T, Top.Largest<T>> create(int numQuantiles) {
- return create(numQuantiles, new Top.Largest<T>());
- }
-
- /**
- * Returns an {@code ApproximateQuantilesCombineFn} that's like
- * this one except that it uses the specified {@code epsilon}
- * value. Does not modify this combiner.
- *
- * <p>See {@link #create(int, Comparator, long,
- * double)} for more information about the meaning of
- * {@code epsilon}.
- */
- public ApproximateQuantilesCombineFn<T, ComparatorT> withEpsilon(double epsilon) {
- return create(numQuantiles, compareFn, maxNumElements, epsilon);
- }
-
- /**
- * Returns an {@code ApproximateQuantilesCombineFn} that's like
- * this one except that it uses the specified {@code maxNumElements}
- * value. Does not modify this combiner.
- *
- * <p>See {@link #create(int, Comparator, long, double)} for more
- * information about the meaning of {@code maxNumElements}.
- */
- public ApproximateQuantilesCombineFn<T, ComparatorT> withMaxInputSize(
- long maxNumElements) {
- return create(numQuantiles, compareFn, maxNumElements, maxNumElements);
- }
-
- /**
- * Creates an approximate quantiles combiner with the given
- * {@code compareFn} and desired number of quantiles. A total of
- * {@code numQuantiles} elements will appear in the output list,
- * including the minimum and maximum.
- *
- * <p>The {@code Comparator} must be {@code Serializable}.
- *
- * <p>The default error bound is {@code epsilon}, which holds as long
- * as the number of elements is less than {@code maxNumElements}.
- * Specifically, if one considers the input as a sorted list x_1, ..., x_N,
- * then the distance between the each exact quantile x_c and its
- * approximation x_c' is bounded by {@code |c - c'| < epsilon * N}.
- * Note that these errors are worst-case scenarios; in practice the accuracy
- * tends to be much better.
- */
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- ApproximateQuantilesCombineFn<T, ComparatorT> create(
- int numQuantiles,
- ComparatorT compareFn,
- long maxNumElements,
- double epsilon) {
- // Compute optimal b and k.
- int b = 2;
- while ((b - 2) * (1 << (b - 2)) < epsilon * maxNumElements) {
- b++;
- }
- b--;
- int k = Math.max(2, (int) Math.ceil(maxNumElements / (1 << (b - 1))));
- return new ApproximateQuantilesCombineFn<T, ComparatorT>(
- numQuantiles, compareFn, k, b, maxNumElements);
- }
-
- @Override
- public QuantileState<T, ComparatorT> createAccumulator() {
- return QuantileState.empty(compareFn, numQuantiles, numBuffers, bufferSize);
- }
-
- @Override
- public Coder<QuantileState<T, ComparatorT>> getAccumulatorCoder(
- CoderRegistry registry, Coder<T> elementCoder) {
- return new QuantileStateCoder<>(compareFn, elementCoder);
- }
- }
-
- /**
- * Compact summarization of a collection on which quantiles can be estimated.
- */
- static class QuantileState<T, ComparatorT extends Comparator<T> & Serializable>
- implements Accumulator<T, QuantileState<T, ComparatorT>, List<T>> {
-
- private ComparatorT compareFn;
- private int numQuantiles;
- private int numBuffers;
- private int bufferSize;
-
- @Nullable
- private T min;
-
- @Nullable
- private T max;
-
- /**
- * The set of buffers, ordered by level from smallest to largest.
- */
- private PriorityQueue<QuantileBuffer<T>> buffers;
-
- /**
- * The algorithm requires that the manipulated buffers always be filled
- * to capacity to perform the collapse operation. This operation can
- * be extended to buffers of varying sizes by introducing the notion of
- * fractional weights, but it's easier to simply combine the remainders
- * from all shards into new, full buffers and then take them into account
- * when computing the final output.
- */
- private List<T> unbufferedElements = Lists.newArrayList();
-
- private QuantileState(
- ComparatorT compareFn,
- int numQuantiles,
- @Nullable T min,
- @Nullable T max,
- int numBuffers,
- int bufferSize,
- Collection<T> unbufferedElements,
- Collection<QuantileBuffer<T>> buffers) {
- this.compareFn = compareFn;
- this.numQuantiles = numQuantiles;
- this.numBuffers = numBuffers;
- this.bufferSize = bufferSize;
- this.buffers = new PriorityQueue<>(numBuffers + 1);
- this.min = min;
- this.max = max;
- this.unbufferedElements.addAll(unbufferedElements);
- this.buffers.addAll(buffers);
- }
-
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- QuantileState<T, ComparatorT> empty(
- ComparatorT compareFn, int numQuantiles, int numBuffers, int bufferSize) {
- return new QuantileState<T, ComparatorT>(
- compareFn,
- numQuantiles,
- null, /* min */
- null, /* max */
- numBuffers,
- bufferSize,
- Collections.<T>emptyList(),
- Collections.<QuantileBuffer<T>>emptyList());
- }
-
- public static <T, ComparatorT extends Comparator<T> & Serializable>
- QuantileState<T, ComparatorT> singleton(
- ComparatorT compareFn, int numQuantiles, T elem, int numBuffers, int bufferSize) {
- return new QuantileState<T, ComparatorT>(
- compareFn,
- numQuantiles,
- elem, /* min */
- elem, /* max */
- numBuffers,
- bufferSize,
- Collections.singletonList(elem),
- Collections.<QuantileBuffer<T>>emptyList());
- }
-
- /**
- * Add a new element to the collection being summarized by this state.
- */
- @Override
- public void addInput(T elem) {
- if (isEmpty()) {
- min = max = elem;
- } else if (compareFn.compare(elem, min) < 0) {
- min = elem;
- } else if (compareFn.compare(elem, max) > 0) {
- max = elem;
- }
- addUnbuffered(elem);
- }
-
- /**
- * Add a new buffer to the unbuffered list, creating a new buffer and
- * collapsing if needed.
- */
- private void addUnbuffered(T elem) {
- unbufferedElements.add(elem);
- if (unbufferedElements.size() == bufferSize) {
- Collections.sort(unbufferedElements, compareFn);
- buffers.add(new QuantileBuffer<T>(unbufferedElements));
- unbufferedElements = Lists.newArrayListWithCapacity(bufferSize);
- collapseIfNeeded();
- }
- }
-
- /**
- * Updates this as if adding all elements seen by other.
- *
- * <p>Note that this ignores the {@code Comparator} of the other {@link QuantileState}. In
- * practice, they should generally be equal, but this method tolerates a mismatch.
- */
- @Override
- public void mergeAccumulator(QuantileState<T, ComparatorT> other) {
- if (other.isEmpty()) {
- return;
- }
- if (min == null || compareFn.compare(other.min, min) < 0) {
- min = other.min;
- }
- if (max == null || compareFn.compare(other.max, max) > 0) {
- max = other.max;
- }
- for (T elem : other.unbufferedElements) {
- addUnbuffered(elem);
- }
- buffers.addAll(other.buffers);
- collapseIfNeeded();
- }
-
- public boolean isEmpty() {
- return unbufferedElements.size() == 0 && buffers.size() == 0;
- }
-
- private void collapseIfNeeded() {
- while (buffers.size() > numBuffers) {
- List<QuantileBuffer<T>> toCollapse = Lists.newArrayList();
- toCollapse.add(buffers.poll());
- toCollapse.add(buffers.poll());
- int minLevel = toCollapse.get(1).level;
- while (!buffers.isEmpty() && buffers.peek().level == minLevel) {
- toCollapse.add(buffers.poll());
- }
- buffers.add(collapse(toCollapse));
- }
- }
-
- private QuantileBuffer<T> collapse(
- Iterable<QuantileBuffer<T>> buffers) {
- int newLevel = 0;
- long newWeight = 0;
- for (QuantileBuffer<T> buffer : buffers) {
- // As presented in the paper, there should always be at least two
- // buffers of the same (minimal) level to collapse, but it is possible
- // to violate this condition when combining buffers from independently
- // computed shards. If they differ we take the max.
- newLevel = Math.max(newLevel, buffer.level + 1);
- newWeight += buffer.weight;
- }
- List<T> newElements =
- interpolate(buffers, bufferSize, newWeight, offset(newWeight));
- return new QuantileBuffer<>(newLevel, newWeight, newElements);
- }
-
- /**
- * If the weight is even, we must round up or down. Alternate between these two options to
- * avoid a bias.
- */
- private long offset(long newWeight) {
- if (newWeight % 2 == 1) {
- return (newWeight + 1) / 2;
- } else {
- offsetJitter = 2 - offsetJitter;
- return (newWeight + offsetJitter) / 2;
- }
- }
-
- /** For alternating between biasing up and down in the above even weight collapse operation. */
- private int offsetJitter = 0;
-
-
- /**
- * Emulates taking the ordered union of all elements in buffers, repeated
- * according to their weight, and picking out the (k * step + offset)-th
- * elements of this list for {@code 0 <= k < count}.
- */
- private List<T> interpolate(Iterable<QuantileBuffer<T>> buffers,
- int count, double step, double offset) {
- List<Iterator<WeightedValue<T>>> iterators = Lists.newArrayList();
- for (QuantileBuffer<T> buffer : buffers) {
- iterators.add(buffer.sizedIterator());
- }
- // Each of the buffers is already sorted by element.
- Iterator<WeightedValue<T>> sorted = Iterators.mergeSorted(
- iterators,
- new Comparator<WeightedValue<T>>() {
- @Override
- public int compare(WeightedValue<T> a, WeightedValue<T> b) {
- return compareFn.compare(a.getValue(), b.getValue());
- }
- });
-
- List<T> newElements = Lists.newArrayListWithCapacity(count);
- WeightedValue<T> weightedElement = sorted.next();
- double current = weightedElement.getWeight();
- for (int j = 0; j < count; j++) {
- double target = j * step + offset;
- while (current <= target && sorted.hasNext()) {
- weightedElement = sorted.next();
- current += weightedElement.getWeight();
- }
- newElements.add(weightedElement.getValue());
- }
- return newElements;
- }
-
- /**
- * Outputs numQuantiles elements consisting of the minimum, maximum, and
- * numQuantiles - 2 evenly spaced intermediate elements.
- *
- * <p>Returns the empty list if no elements have been added.
- */
- @Override
- public List<T> extractOutput() {
- if (isEmpty()) {
- return Lists.newArrayList();
- }
- long totalCount = unbufferedElements.size();
- for (QuantileBuffer<T> buffer : buffers) {
- totalCount += bufferSize * buffer.weight;
- }
- List<QuantileBuffer<T>> all = Lists.newArrayList(buffers);
- if (!unbufferedElements.isEmpty()) {
- Collections.sort(unbufferedElements, compareFn);
- all.add(new QuantileBuffer<>(unbufferedElements));
- }
- double step = 1.0 * totalCount / (numQuantiles - 1);
- double offset = (1.0 * totalCount - 1) / (numQuantiles - 1);
- List<T> quantiles = interpolate(all, numQuantiles - 2, step, offset);
- quantiles.add(0, min);
- quantiles.add(max);
- return quantiles;
- }
- }
-
- /**
- * A single buffer in the sense of the referenced algorithm.
- */
- private static class QuantileBuffer<T> implements Comparable<QuantileBuffer<T>> {
- private int level;
- private long weight;
- private List<T> elements;
-
- public QuantileBuffer(List<T> elements) {
- this(0, 1, elements);
- }
-
- public QuantileBuffer(int level, long weight, List<T> elements) {
- this.level = level;
- this.weight = weight;
- this.elements = elements;
- }
-
- @Override
- public int compareTo(QuantileBuffer<T> other) {
- return this.level - other.level;
- }
-
- @Override
- public String toString() {
- return "QuantileBuffer["
- + "level=" + level
- + ", weight="
- + weight + ", elements=" + elements + "]";
- }
-
- public Iterator<WeightedValue<T>> sizedIterator() {
- return new UnmodifiableIterator<WeightedValue<T>>() {
- Iterator<T> iter = elements.iterator();
- @Override
- public boolean hasNext() {
- return iter.hasNext();
- }
- @Override public WeightedValue<T> next() {
- return WeightedValue.of(iter.next(), weight);
- }
- };
- }
- }
-
- /**
- * Coder for QuantileState.
- */
- private static class QuantileStateCoder<T, ComparatorT extends Comparator<T> & Serializable>
- extends CustomCoder<QuantileState<T, ComparatorT>> {
- private final ComparatorT compareFn;
- private final Coder<T> elementCoder;
- private final Coder<List<T>> elementListCoder;
- private final Coder<Integer> intCoder = BigEndianIntegerCoder.of();
-
- public QuantileStateCoder(ComparatorT compareFn, Coder<T> elementCoder) {
- this.compareFn = compareFn;
- this.elementCoder = elementCoder;
- this.elementListCoder = ListCoder.of(elementCoder);
- }
-
- @Override
- public void encode(
- QuantileState<T, ComparatorT> state, OutputStream outStream, Coder.Context context)
- throws CoderException, IOException {
- Coder.Context nestedContext = context.nested();
- intCoder.encode(state.numQuantiles, outStream, nestedContext);
- intCoder.encode(state.bufferSize, outStream, nestedContext);
- elementCoder.encode(state.min, outStream, nestedContext);
- elementCoder.encode(state.max, outStream, nestedContext);
- elementListCoder.encode(
- state.unbufferedElements, outStream, nestedContext);
- BigEndianIntegerCoder.of().encode(
- state.buffers.size(), outStream, nestedContext);
- for (QuantileBuffer<T> buffer : state.buffers) {
- encodeBuffer(buffer, outStream, nestedContext);
- }
- }
-
- @Override
- public QuantileState<T, ComparatorT> decode(InputStream inStream, Coder.Context context)
- throws CoderException, IOException {
- Coder.Context nestedContext = context.nested();
- int numQuantiles = intCoder.decode(inStream, nestedContext);
- int bufferSize = intCoder.decode(inStream, nestedContext);
- T min = elementCoder.decode(inStream, nestedContext);
- T max = elementCoder.decode(inStream, nestedContext);
- List<T> unbufferedElements =
- elementListCoder.decode(inStream, nestedContext);
- int numBuffers =
- BigEndianIntegerCoder.of().decode(inStream, nestedContext);
- List<QuantileBuffer<T>> buffers = new ArrayList<>(numBuffers);
- for (int i = 0; i < numBuffers; i++) {
- buffers.add(decodeBuffer(inStream, nestedContext));
- }
- return new QuantileState<T, ComparatorT>(
- compareFn, numQuantiles, min, max, numBuffers, bufferSize, unbufferedElements, buffers);
- }
-
- private void encodeBuffer(
- QuantileBuffer<T> buffer, OutputStream outStream, Coder.Context context)
- throws CoderException, IOException {
- DataOutputStream outData = new DataOutputStream(outStream);
- outData.writeInt(buffer.level);
- outData.writeLong(buffer.weight);
- elementListCoder.encode(buffer.elements, outStream, context);
- }
-
- private QuantileBuffer<T> decodeBuffer(
- InputStream inStream, Coder.Context context)
- throws IOException, CoderException {
- DataInputStream inData = new DataInputStream(inStream);
- return new QuantileBuffer<>(
- inData.readInt(),
- inData.readLong(),
- elementListCoder.decode(inStream, context));
- }
-
- /**
- * Notifies ElementByteSizeObserver about the byte size of the
- * encoded value using this coder.
- */
- @Override
- public void registerByteSizeObserver(
- QuantileState<T, ComparatorT> state,
- ElementByteSizeObserver observer,
- Coder.Context context)
- throws Exception {
- Coder.Context nestedContext = context.nested();
- elementCoder.registerByteSizeObserver(
- state.min, observer, nestedContext);
- elementCoder.registerByteSizeObserver(
- state.max, observer, nestedContext);
- elementListCoder.registerByteSizeObserver(
- state.unbufferedElements, observer, nestedContext);
-
- BigEndianIntegerCoder.of().registerByteSizeObserver(
- state.buffers.size(), observer, nestedContext);
- for (QuantileBuffer<T> buffer : state.buffers) {
- observer.update(4L + 8);
-
- elementListCoder.registerByteSizeObserver(
- buffer.elements, observer, nestedContext);
- }
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- verifyDeterministic(
- "QuantileState.ElementCoder must be deterministic",
- elementCoder);
- verifyDeterministic(
- "QuantileState.ElementListCoder must be deterministic",
- elementListCoder);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateUnique.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateUnique.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateUnique.java
deleted file mode 100644
index 3c936a2..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/ApproximateUnique.java
+++ /dev/null
@@ -1,419 +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.Coder.Context;
-import com.google.cloud.dataflow.sdk.coders.CoderException;
-import com.google.cloud.dataflow.sdk.coders.CoderRegistry;
-import com.google.cloud.dataflow.sdk.coders.KvCoder;
-import com.google.cloud.dataflow.sdk.coders.SerializableCoder;
-import com.google.cloud.dataflow.sdk.transforms.Combine.CombineFn;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.common.hash.Hashing;
-import com.google.common.hash.HashingOutputStream;
-import com.google.common.io.ByteStreams;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
-import java.util.PriorityQueue;
-
-/**
- * {@code PTransform}s for estimating the number of distinct elements
- * in a {@code PCollection}, or the number of distinct values
- * associated with each key in a {@code PCollection} of {@code KV}s.
- */
-public class ApproximateUnique {
-
- /**
- * Returns a {@code PTransform} that takes a {@code PCollection<T>}
- * and returns a {@code PCollection<Long>} containing a single value
- * that is an estimate of the number of distinct elements in the
- * input {@code PCollection}.
- *
- * <p>The {@code sampleSize} parameter controls the estimation
- * error. The error is about {@code 2 / sqrt(sampleSize)}, so for
- * {@code ApproximateUnique.globally(10000)} the estimation error is
- * about 2%. Similarly, for {@code ApproximateUnique.of(16)} the
- * estimation error is about 50%. If there are fewer than
- * {@code sampleSize} distinct elements then the returned result
- * will be exact with extremely high probability (the chance of a
- * hash collision is about {@code sampleSize^2 / 2^65}).
- *
- * <p>This transform approximates the number of elements in a set
- * by computing the top {@code sampleSize} hash values, and using
- * that to extrapolate the size of the entire set of hash values by
- * assuming the rest of the hash values are as densely distributed
- * as the top {@code sampleSize}.
- *
- * <p>See also {@link #globally(double)}.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<String> pc = ...;
- * PCollection<Long> approxNumDistinct =
- * pc.apply(ApproximateUnique.<String>globally(1000));
- * } </pre>
- *
- * @param <T> the type of the elements in the input {@code PCollection}
- * @param sampleSize the number of entries in the statistical
- * sample; the higher this number, the more accurate the
- * estimate will be; should be {@code >= 16}
- * @throws IllegalArgumentException if the {@code sampleSize}
- * argument is too small
- */
- public static <T> Globally<T> globally(int sampleSize) {
- return new Globally<>(sampleSize);
- }
-
- /**
- * Like {@link #globally(int)}, but specifies the desired maximum
- * estimation error instead of the sample size.
- *
- * @param <T> the type of the elements in the input {@code PCollection}
- * @param maximumEstimationError the maximum estimation error, which
- * should be in the range {@code [0.01, 0.5]}
- * @throws IllegalArgumentException if the
- * {@code maximumEstimationError} argument is out of range
- */
- public static <T> Globally<T> globally(double maximumEstimationError) {
- return new Globally<>(maximumEstimationError);
- }
-
- /**
- * Returns a {@code PTransform} that takes a
- * {@code PCollection<KV<K, V>>} and returns a
- * {@code PCollection<KV<K, Long>>} that contains an output element
- * mapping each distinct key in the input {@code PCollection} to an
- * estimate of the number of distinct values associated with that
- * key in the input {@code PCollection}.
- *
- * <p>See {@link #globally(int)} for an explanation of the
- * {@code sampleSize} parameter. A separate sampling is computed
- * for each distinct key of the input.
- *
- * <p>See also {@link #perKey(double)}.
- *
- * <p>Example of use:
- * <pre> {@code
- * PCollection<KV<Integer, String>> pc = ...;
- * PCollection<KV<Integer, Long>> approxNumDistinctPerKey =
- * pc.apply(ApproximateUnique.<Integer, String>perKey(1000));
- * } </pre>
- *
- * @param <K> the type of the keys in the input and output
- * {@code PCollection}s
- * @param <V> the type of the values in the input {@code PCollection}
- * @param sampleSize the number of entries in the statistical
- * sample; the higher this number, the more accurate the
- * estimate will be; should be {@code >= 16}
- * @throws IllegalArgumentException if the {@code sampleSize}
- * argument is too small
- */
- public static <K, V> PerKey<K, V> perKey(int sampleSize) {
- return new PerKey<>(sampleSize);
- }
-
- /**
- * Like {@link #perKey(int)}, but specifies the desired maximum
- * estimation error instead of the sample size.
- *
- * @param <K> the type of the keys in the input and output
- * {@code PCollection}s
- * @param <V> the type of the values in the input {@code PCollection}
- * @param maximumEstimationError the maximum estimation error, which
- * should be in the range {@code [0.01, 0.5]}
- * @throws IllegalArgumentException if the
- * {@code maximumEstimationError} argument is out of range
- */
- public static <K, V> PerKey<K, V> perKey(double maximumEstimationError) {
- return new PerKey<>(maximumEstimationError);
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * {@code PTransform} for estimating the number of distinct elements
- * in a {@code PCollection}.
- *
- * @param <T> the type of the elements in the input {@code PCollection}
- */
- static class Globally<T> extends PTransform<PCollection<T>, PCollection<Long>> {
-
- /**
- * The number of entries in the statistical sample; the higher this number,
- * the more accurate the estimate will be.
- */
- private final long sampleSize;
-
- /**
- * @see ApproximateUnique#globally(int)
- */
- public Globally(int sampleSize) {
- if (sampleSize < 16) {
- throw new IllegalArgumentException(
- "ApproximateUnique needs a sampleSize "
- + ">= 16 for an estimation error <= 50%. "
- + "In general, the estimation "
- + "error is about 2 / sqrt(sampleSize).");
- }
- this.sampleSize = sampleSize;
- }
-
- /**
- * @see ApproximateUnique#globally(double)
- */
- public Globally(double maximumEstimationError) {
- if (maximumEstimationError < 0.01 || maximumEstimationError > 0.5) {
- throw new IllegalArgumentException(
- "ApproximateUnique needs an "
- + "estimation error between 1% (0.01) and 50% (0.5).");
- }
- this.sampleSize = sampleSizeFromEstimationError(maximumEstimationError);
- }
-
- @Override
- public PCollection<Long> apply(PCollection<T> input) {
- Coder<T> coder = input.getCoder();
- return input.apply(
- Combine.globally(
- new ApproximateUniqueCombineFn<>(sampleSize, coder)));
- }
- }
-
- /**
- * {@code PTransform} for estimating the number of distinct values
- * associated with each key in a {@code PCollection} of {@code KV}s.
- *
- * @param <K> the type of the keys in the input and output
- * {@code PCollection}s
- * @param <V> the type of the values in the input {@code PCollection}
- */
- static class PerKey<K, V>
- extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Long>>> {
-
- private final long sampleSize;
-
- /**
- * @see ApproximateUnique#perKey(int)
- */
- public PerKey(int sampleSize) {
- if (sampleSize < 16) {
- throw new IllegalArgumentException(
- "ApproximateUnique needs a "
- + "sampleSize >= 16 for an estimation error <= 50%. In general, "
- + "the estimation error is about 2 / sqrt(sampleSize).");
- }
- this.sampleSize = sampleSize;
- }
-
- /**
- * @see ApproximateUnique#perKey(double)
- */
- public PerKey(double estimationError) {
- if (estimationError < 0.01 || estimationError > 0.5) {
- throw new IllegalArgumentException(
- "ApproximateUnique.PerKey needs an "
- + "estimation error between 1% (0.01) and 50% (0.5).");
- }
- this.sampleSize = sampleSizeFromEstimationError(estimationError);
- }
-
- @Override
- public PCollection<KV<K, Long>> apply(PCollection<KV<K, V>> input) {
- Coder<KV<K, V>> inputCoder = input.getCoder();
- if (!(inputCoder instanceof KvCoder)) {
- throw new IllegalStateException(
- "ApproximateUnique.PerKey requires its input to use KvCoder");
- }
- @SuppressWarnings("unchecked")
- final Coder<V> coder = ((KvCoder<K, V>) inputCoder).getValueCoder();
-
- return input.apply(
- Combine.perKey(new ApproximateUniqueCombineFn<>(
- sampleSize, coder).<K>asKeyedFn()));
- }
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * {@code CombineFn} that computes an estimate of the number of
- * distinct values that were combined.
- *
- * <p>Hashes input elements, computes the top {@code sampleSize}
- * hash values, and uses those to extrapolate the size of the entire
- * set of hash values by assuming the rest of the hash values are as
- * densely distributed as the top {@code sampleSize}.
- *
- * <p>Used to implement
- * {@link #globally(int) ApproximatUnique.globally(...)} and
- * {@link #perKey(int) ApproximatUnique.perKey(...)}.
- *
- * @param <T> the type of the values being combined
- */
- public static class ApproximateUniqueCombineFn<T> extends
- CombineFn<T, ApproximateUniqueCombineFn.LargestUnique, Long> {
-
- /**
- * The size of the space of hashes returned by the hash function.
- */
- static final double HASH_SPACE_SIZE =
- Long.MAX_VALUE - (double) Long.MIN_VALUE;
-
- /**
- * A heap utility class to efficiently track the largest added elements.
- */
- public static class LargestUnique implements Serializable {
- private PriorityQueue<Long> heap = new PriorityQueue<>();
- private final long sampleSize;
-
- /**
- * Creates a heap to track the largest {@code sampleSize} elements.
- *
- * @param sampleSize the size of the heap
- */
- public LargestUnique(long sampleSize) {
- this.sampleSize = sampleSize;
- }
-
- /**
- * Adds a value to the heap, returning whether the value is (large enough
- * to be) in the heap.
- */
- public boolean add(Long value) {
- if (heap.contains(value)) {
- return true;
- } else if (heap.size() < sampleSize) {
- heap.add(value);
- return true;
- } else if (value > heap.element()) {
- heap.remove();
- heap.add(value);
- return true;
- } else {
- return false;
- }
- }
-
- /**
- * Returns the values in the heap, ordered largest to smallest.
- */
- public List<Long> extractOrderedList() {
- // The only way to extract the order from the heap is element-by-element
- // from smallest to largest.
- Long[] array = new Long[heap.size()];
- for (int i = heap.size() - 1; i >= 0; i--) {
- array[i] = heap.remove();
- }
- return Arrays.asList(array);
- }
- }
-
- private final long sampleSize;
- private final Coder<T> coder;
-
- public ApproximateUniqueCombineFn(long sampleSize, Coder<T> coder) {
- this.sampleSize = sampleSize;
- this.coder = coder;
- }
-
- @Override
- public LargestUnique createAccumulator() {
- return new LargestUnique(sampleSize);
- }
-
- @Override
- public LargestUnique addInput(LargestUnique heap, T input) {
- try {
- heap.add(hash(input, coder));
- return heap;
- } catch (Throwable e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public LargestUnique mergeAccumulators(Iterable<LargestUnique> heaps) {
- Iterator<LargestUnique> iterator = heaps.iterator();
- LargestUnique heap = iterator.next();
- while (iterator.hasNext()) {
- List<Long> largestHashes = iterator.next().extractOrderedList();
- for (long hash : largestHashes) {
- if (!heap.add(hash)) {
- break; // The remainder of this list is all smaller.
- }
- }
- }
- return heap;
- }
-
- @Override
- public Long extractOutput(LargestUnique heap) {
- List<Long> largestHashes = heap.extractOrderedList();
- if (largestHashes.size() < sampleSize) {
- return (long) largestHashes.size();
- } else {
- long smallestSampleHash = largestHashes.get(largestHashes.size() - 1);
- double sampleSpaceSize = Long.MAX_VALUE - (double) smallestSampleHash;
- // This formula takes into account the possibility of hash collisions,
- // which become more likely than not for 2^32 distinct elements.
- // Note that log(1+x) ~ x for small x, so for sampleSize << maxHash
- // log(1 - sampleSize/sampleSpace) / log(1 - 1/sampleSpace) ~ sampleSize
- // and hence estimate ~ sampleSize * HASH_SPACE_SIZE / sampleSpace
- // as one would expect.
- double estimate = Math.log1p(-sampleSize / sampleSpaceSize)
- / Math.log1p(-1 / sampleSpaceSize)
- * HASH_SPACE_SIZE / sampleSpaceSize;
- return Math.round(estimate);
- }
- }
-
- @Override
- public Coder<LargestUnique> getAccumulatorCoder(CoderRegistry registry,
- Coder<T> inputCoder) {
- return SerializableCoder.of(LargestUnique.class);
- }
-
- /**
- * Encodes the given element using the given coder and hashes the encoding.
- */
- static <T> long hash(T element, Coder<T> coder) throws CoderException, IOException {
- try (HashingOutputStream stream =
- new HashingOutputStream(Hashing.murmur3_128(), ByteStreams.nullOutputStream())) {
- coder.encode(element, stream, Context.OUTER);
- return stream.hash().asLong();
- }
- }
- }
-
- /**
- * Computes the sampleSize based on the desired estimation error.
- *
- * @param estimationError should be bounded by [0.01, 0.5]
- * @return the sample size needed for the desired estimation error
- */
- static long sampleSizeFromEstimationError(double estimationError) {
- return Math.round(Math.ceil(4.0 / Math.pow(estimationError, 2.0)));
- }
-}