def test_pardo_side_input(self):
    p = df.Pipeline('DirectPipelineRunner')
    words = p | df.Create('start', ['a', 'bb', 'ccc', 'dddd'])

    # [START model_pardo_side_input]
    # Callable takes additional arguments.
    def filter_using_length(word, lower_bound, upper_bound=float('inf')):
      if lower_bound <= len(word) <= upper_bound:
        yield word

    # Construct a deferred side input.
    avg_word_len = words | df.Map(len) | df.CombineGlobally(df.combiners.MeanCombineFn())

    # Call with explicit side inputs.
    small_words = words | df.FlatMap('small', filter_using_length, 0, 3)

    # A single deferred side input.
    larger_than_average = words | df.FlatMap('large',
                                             filter_using_length,
                                             lower_bound=pvalue.AsSingleton(avg_word_len))

    # Mix and match.
    small_but_nontrivial = words | df.FlatMap(filter_using_length,
                                              lower_bound=2,
                                              upper_bound=pvalue.AsSingleton(avg_word_len))
    # [END model_pardo_side_input]

    df.assert_that(small_words, df.equal_to(['a', 'bb', 'ccc']))
    df.assert_that(larger_than_average, df.equal_to(['ccc', 'dddd']),
                   label='larger_than_average')
    df.assert_that(small_but_nontrivial, df.equal_to(['bb']),
                   label='small_but_not_trivial')
    p.run()
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/examples/snippets/snippets_test.py/ParDoTest.test_pardo_side_input/original