original_docstring: Converts input source to TF example splits based on configs.
  
      Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
      for converting input split to TF Examples. Overriding this
      'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
      custom spliting logic.
  
      Args:
        pipeline: beam pipeline.
        input_dict: Input dict from input key to a list of Artifacts. Depends on
          detailed example gen implementation.
        exec_properties: A dict of execution properties. Depends on detailed
          example gen implementation.
          - input: JSON string of example_gen_pb2.Input instance, providing input
            configuration.
          - output: JSON string of example_gen_pb2.Output instance, providing
            output configuration.
  
      Returns:
        Dict of beam PCollection with split name as key, each PCollection is a
        single output split that contains serialized TF Examples.
      
docstring: Converts input source to TF example splits based on configs.
  Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
  for converting input split to TF Examples. Overriding this
  'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
  custom spliting logic.
docstring_tokens: Converts input source to TF example splits based on configs . Custom ExampleGen executor should provide GetInputSourceToExamplePTransform for converting input split to TF Examples . Overriding this ' GenerateExamplesByBeam ' method instead if complex logic is need e . g . custom spliting logic .
code: def GenerateExamplesByBeam(self, pipeline: beam.Pipeline,
                               input_dict: Dict[Text, List[types.TfxArtifact]],
                               exec_properties: Dict[Text, Any]
                              ) -> Dict[Text, beam.pvalue.PCollection]:
      """Converts input source to TF example splits based on configs.
  
      Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
      for converting input split to TF Examples. Overriding this
      'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
      custom spliting logic.
  
      Args:
        pipeline: beam pipeline.
        input_dict: Input dict from input key to a list of Artifacts. Depends on
          detailed example gen implementation.
        exec_properties: A dict of execution properties. Depends on detailed
          example gen implementation.
          - input: JSON string of example_gen_pb2.Input instance, providing input
            configuration.
          - output: JSON string of example_gen_pb2.Output instance, providing
            output configuration.
  
      Returns:
        Dict of beam PCollection with split name as key, each PCollection is a
        single output split that contains serialized TF Examples.
      """
      # Get input split information.
      input_config = example_gen_pb2.Input()
      json_format.Parse(exec_properties['input'], input_config)
      # Get output split information.
      output_config = example_gen_pb2.Output()
      json_format.Parse(exec_properties['output'], output_config)
      # Get output split names.
      split_names = utils.generate_output_split_names(input_config, output_config)
  
      example_splits = []
      input_to_example = self.GetInputSourceToExamplePTransform()
      if output_config.split_config.splits:
        # Use output splits, input must have only one split.
        assert len(
            input_config.splits
        ) == 1, 'input must have only one split when output split is specified.'
        # Calculate split buckets.
        buckets = []
        total_buckets = 0
        for split in output_config.split_config.splits:
          total_buckets += split.hash_buckets
          buckets.append(total_buckets)
        example_splits = (
            pipeline
            | 'InputToSerializedExample' >> _InputToSerializedExample(  # pylint: disable=no-value-for-parameter
                input_to_example, input_dict, exec_properties,
                input_config.splits[0].pattern)
            | 'SplitData' >> beam.Partition(_PartitionFn, len(buckets), buckets))
      else:
        # Use input splits.
        for split in input_config.splits:
          examples = (
              pipeline
              | 'InputToSerializedExample' + split.name >>
              _InputToSerializedExample(  # pylint: disable=no-value-for-parameter
                  input_to_example, input_dict, exec_properties, split.pattern))
          example_splits.append(examples)
  
      result = {}
      for index, example_split in enumerate(example_splits):
        result[split_names[index]] = example_split
      return result
code_tokens: def GenerateExamplesByBeam ( self , pipeline : beam . Pipeline , input_dict : Dict [ Text , List [ types . TfxArtifact ] ] , exec_properties : Dict [ Text , Any ] ) -> Dict [ Text , beam . pvalue . PCollection ] : input_config = example_gen_pb2 . Input ( ) json_format . Parse ( exec_properties [ 'input' ] , input_config ) output_config = example_gen_pb2 . Output ( ) json_format . Parse ( exec_properties [ 'output' ] , output_config ) split_names = utils . generate_output_split_names ( input_config , output_config ) example_splits = [ ] input_to_example = self . GetInputSourceToExamplePTransform ( ) if output_config . split_config . splits : assert len ( input_config . splits ) == 1 , 'input must have only one split when output split is specified.' buckets = [ ] total_buckets = 0 for split in output_config . split_config . splits : total_buckets += split . hash_buckets buckets . append ( total_buckets ) example_splits = ( pipeline | 'InputToSerializedExample' >> _InputToSerializedExample ( input_to_example , input_dict , exec_properties , input_config . splits [ 0 ] . pattern ) | 'SplitData' >> beam . Partition ( _PartitionFn , len ( buckets ) , buckets ) ) else : for split in input_config . splits : examples = ( pipeline | 'InputToSerializedExample' + split . name >> _InputToSerializedExample ( input_to_example , input_dict , exec_properties , split . pattern ) ) example_splits . append ( examples ) result = { } for index , example_split in enumerate ( example_splits ) : result [ split_names [ index ] ] = example_split return result
short_docstring: Converts input source to TF example splits based on configs.
short_docstring_tokens: Converts input source to TF example splits based on configs .
comment: """Converts input source to TF example splits based on configs.
  
      Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
      for converting input split to TF Examples. Overriding this
      'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
      custom spliting logic.
  
      Args:
        pipeline: beam pipeline.
        input_dict: Input dict from input key to a list of Artifacts. Depends on
          detailed example gen implementation.
        exec_properties: A dict of execution properties. Depends on detailed
          example gen implementation.
          - input: JSON string of example_gen_pb2.Input instance, providing input
            configuration.
          - output: JSON string of example_gen_pb2.Output instance, providing
            output configuration.
  
      Returns:
        Dict of beam PCollection with split name as key, each PCollection is a
        single output split that contains serialized TF Examples.
      """$SEP$# Get input split information.$SEP$# Get output split information.$SEP$# Get output split names.$SEP$# Use output splits, input must have only one split.$SEP$# Calculate split buckets.$SEP$# pylint: disable=no-value-for-parameter$SEP$# Use input splits.$SEP$# pylint: disable=no-value-for-parameter