def test_top(self):
    pipeline = Pipeline('DirectPipelineRunner')

    # A parameter we'll be sharing with a custom comparator.
    names = {0: 'zo',
             1: 'one',
             2: 'twoo',
             3: 'three',
             5: 'fiiive',
             6: 'sssssix',
             9: 'nniiinne'}

    # First for global combines.
    pcoll = pipeline | Create('start', [6, 3, 1, 1, 9, 1, 5, 2, 0, 6])
    result_top = pcoll | combine.Top.Largest('top', 5)
    result_bot = pcoll | combine.Top.Smallest('bot', 4)
    result_cmp = pcoll | combine.Top.Of(
        'cmp',
        6,
        lambda a, b, names: len(names[a]) < len(names[b]),
        names)  # Note parameter passed to comparator.
    assert_that(result_top, equal_to([[9, 6, 6, 5, 3]]), label='assert:top')
    assert_that(result_bot, equal_to([[0, 1, 1, 1]]), label='assert:bot')
    assert_that(result_cmp, equal_to([[9, 6, 6, 5, 3, 2]]), label='assert:cmp')

    # Again for per-key combines.
    pcoll = pipeline | Create(
        'start-perkey', [('a', x) for x in [6, 3, 1, 1, 9, 1, 5, 2, 0, 6]])
    result_key_top = pcoll | combine.Top.LargestPerKey('top-perkey', 5)
    result_key_bot = pcoll | combine.Top.SmallestPerKey('bot-perkey', 4)
    result_key_cmp = pcoll | combine.Top.PerKey(
        'cmp-perkey',
        6,
        lambda a, b, names: len(names[b]) < len(names[a]),
        names)  # Note parameter passed to comparator.
    assert_that(result_key_top, equal_to([('a', [9, 6, 6, 5, 3])]),
                label='key:top')
    assert_that(result_key_bot, equal_to([('a', [0, 1, 1, 1])]),
                label='key:bot')
    assert_that(result_key_cmp, equal_to([('a', [9, 6, 6, 5, 3, 2])]),
                label='key:cmp')
    pipeline.run()
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/transforms/combiners_test.py/CombineTest.test_top/swapped operands of `<`