def test_pardo_with_side_outputs(self):
    # [START model_pardo_emitting_values_on_side_outputs]
    class ProcessWords(df.DoFn):

      def process(self, context, cutoff_length, marker):
        if len(context.element) <= cutoff_length:
          # Emit this short word to the main output.
          yield context.element
        else:
          # Emit this word's long length to a side output.
          yield pvalue.SideOutputValue(
              'above_cutoff_lengths', len(context.element))
        if context.element.startswith(marker):
          # Emit this word to a different side output.
          yield pvalue.SideOutputValue('marked strings', context.element)
    # [END model_pardo_emitting_values_on_side_outputs]

    words = ['a', 'an', 'the', 'music', 'xyz']

    # [START model_pardo_with_side_outputs]
    results = (words | df.ParDo(ProcessWords(), cutoff_length=2, marker='x')
                         .with_outputs('above_cutoff_lengths',
                                       'marked strings',
                                       main='below_cutoff_strings'))
    below = results.below_cutoff_strings
    above = results.above_cutoff_lengths
    marked = results['marked strings']  # indexing works as well
    # [END model_pardo_with_side_outputs]

    self.assertEqual({'a', 'an'}, set(below))
    self.assertEqual({3, 5}, set(above))
    self.assertEqual({'xyz'}, set(marked))

    # [START model_pardo_with_side_outputs_iter]
    below, above, marked = (words | df.ParDo(ProcessWords(), cutoff_length=2, marker='x')
                                      .with_outputs('above_cutoff_lengths',
                                                    'marked strings',
                                                    main='below_cutoff_strings'))
    # [END model_pardo_with_side_outputs_iter]

    self.assertEqual({'a', 'an'}, set(below))
    self.assertEqual({3, 5}, set(above))
    self.assertEqual({'xyz'}, set(marked))
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/examples/snippets/snippets_test.py/ParDoTest.test_pardo_with_side_outputs/original