### Question:
ReservoirSampling { public static <T, N extends Number> PCollection<T> weightedSample( PCollection<Pair<T, N>> input, int sampleSize) { return weightedSample(input, sampleSize, null); } private  ReservoirSampling(); static PCollection<T> sample(
      PCollection<T> input,
      int sampleSize); static PCollection<T> sample(
      PCollection<T> input,
      int sampleSize,
      RandomGenerator random); static PCollection<T> weightedSample(
      PCollection<Pair<T, N>> input,
      int sampleSize); static PCollection<T> weightedSample(
      PCollection<Pair<T, N>> input,
      int sampleSize,
      RandomGenerator random); static PTable<K, T> groupedWeightedSample(
      PTable<K, Pair<T, N>> input,
      int sampleSize); static PTable<K, T> groupedWeightedSample(
      PTable<K, Pair<T, N>> input,
      int sampleSize,
      RandomGenerator random);  }### Answer:
@Test public void testWRS() throws Exception { Map<String, Integer> histogram = Maps.newHashMap(); RandomGenerator r = RandomManager.getRandom(); for (int i = 0; i < 100; i++) { PCollection<String> sample = ReservoirSampling.weightedSample(VALUES, 2, r); for (String s : sample.materialize()) { if (histogram.containsKey(s)) { histogram.put(s, 1 + histogram.get(s)); } else { histogram.put(s, 1); } } } Map<String, Integer> expected = ImmutableMap.of( "foo", 48, "bar", 80, "baz", 32, "biz", 40); assertEquals(expected, histogram); }