def test_sample(self):

    # First test global samples (lots of them).
    for ix in xrange(300):
      pipeline = Pipeline('DirectPipelineRunner')
      pcoll = pipeline | Create('start', [1, 1, 2, 2])
      result = pcoll | combine.Sample.FixedSizeGlobally('sample-%d' % ix, 3)
      def matcher():
        def match(actual):
          # There is always exactly one result.
          equal_to([1])([len(actual)])
          # There are always exactly three samples in the result.
          equal_to([3])([len(actual[0])])
          # Sampling is without replacement.
          num_ones = sum(1 for x in actual[0] if x == 1)
          num_twos = sum(1 for x in actual[0] if x == 2)
          equal_to([1, 2])([num_ones, num_twos])
        return match
      assert_that(result, matcher())
      pipeline.run()

    # Now test per-key samples.
    pipeline = Pipeline('DirectPipelineRunner')
    pcoll = pipeline | Create(
        'start-perkey',
        sum(([(i, 1), (i, 1), (i, 2), (i, 2)] for i in xrange(300)), []))
    result = pcoll | combine.Sample.FixedSizePerKey('sample', 3)
    def matcher():
      def match(actual):
        for _, samples in actual:
          equal_to([3])([len(samples)])
          num_ones = sum(1 for x in samples if x == 1)
          num_twos = sum(1 for x in samples if x == 2)
          equal_to([1, 2])([num_ones, num_twos])
      return match
    assert_that(result, matcher())
    pipeline.run()
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/transforms/combiners_test.py/CombineTest.test_sample/original