(pipeline,
                        input_iterator,
                        output_dir,
                        output_file_base=None)
Runs the a pipeline on a data source and writes to a directory.

  Run the pipeline on each input from the iterator one at a time.
  A file will be written to `output_dir` for each dataset name specified
  by the pipeline. pipeline.transform is called on each input and the
  results are aggregated into their correct datasets.

  The output type or types given by `pipeline.output_type` must be protocol
  buffers or objects that have a SerializeToString method.

  Args:
    pipeline: A Pipeline instance. `pipeline.output_type` must be a protocol
        buffer or a dictionary mapping names to protocol buffers.
    input_iterator: Iterates over the input data. Items returned by it are fed
        directly into the pipeline's `transform` method.
    output_dir: Path to directory where datasets will be written. Each dataset
        is a file whose name contains the pipeline's dataset name. If the
        directory does not exist, it will be created.
    output_file_base: An optional string prefix for all datasets output by this
        run. The prefix will also be followed by an underscore.

  Raises:
    ValueError: If any of `pipeline`'s output types do not have a
        SerializeToString method.
Runs the a pipeline on a data source and writes to a directory.
Runs the a pipeline on a data source and writes to a directory .
def run_pipeline_serial(pipeline,
                        input_iterator,
                        output_dir,
                        output_file_base=None):
  """Runs the a pipeline on a data source and writes to a directory.

  Run the pipeline on each input from the iterator one at a time.
  A file will be written to `output_dir` for each dataset name specified
  by the pipeline. pipeline.transform is called on each input and the
  results are aggregated into their correct datasets.

  The output type or types given by `pipeline.output_type` must be protocol
  buffers or objects that have a SerializeToString method.

  Args:
    pipeline: A Pipeline instance. `pipeline.output_type` must be a protocol
        buffer or a dictionary mapping names to protocol buffers.
    input_iterator: Iterates over the input data. Items returned by it are fed
        directly into the pipeline's `transform` method.
    output_dir: Path to directory where datasets will be written. Each dataset
        is a file whose name contains the pipeline's dataset name. If the
        directory does not exist, it will be created.
    output_file_base: An optional string prefix for all datasets output by this
        run. The prefix will also be followed by an underscore.

  Raises:
    ValueError: If any of `pipeline`'s output types do not have a
        SerializeToString method.
  """
  if isinstance(pipeline.output_type, dict):
    for name, type_ in pipeline.output_type.items():
      if not hasattr(type_, 'SerializeToString'):
        raise ValueError(
            'Pipeline output "%s" does not have method SerializeToString. '
            'Output type = %s' % (name, pipeline.output_type))
  else:
    if not hasattr(pipeline.output_type, 'SerializeToString'):
      raise ValueError(
          'Pipeline output type %s does not have method SerializeToString.'
          % pipeline.output_type)

  if not tf.gfile.Exists(output_dir):
    tf.gfile.MakeDirs(output_dir)

  output_names = pipeline.output_type_as_dict.keys()

  if output_file_base is None:
    output_paths = [os.path.join(output_dir, name + '.tfrecord')
                    for name in output_names]
  else:
    output_paths = [os.path.join(output_dir,
                                 '%s_%s.tfrecord' % (output_file_base, name))
                    for name in output_names]

  writers = dict((name, tf.python_io.TFRecordWriter(path))
                 for name, path in zip(output_names, output_paths))

  total_inputs = 0
  total_outputs = 0
  stats = []
  for input_ in input_iterator:
    total_inputs += 1
    for name, outputs in _guarantee_dict(pipeline.transform(input_),
                                         list(output_names)[0]).items():
      for output in outputs:  # pylint:disable=not-an-iterable
        writers[name].write(output.SerializeToString())
      total_outputs += len(outputs)
    stats = statistics.merge_statistics(stats + pipeline.get_stats())
    if total_inputs % 500 == 0:
      tf.logging.info('Processed %d inputs so far. Produced %d outputs.',
                      total_inputs, total_outputs)
      statistics.log_statistics_list(stats, tf.logging.info)
  tf.logging.info('\n\nCompleted.\n')
  tf.logging.info('Processed %d inputs total. Produced %d outputs.',
                  total_inputs, total_outputs)
  statistics.log_statistics_list(stats, tf.logging.info)
def run_pipeline_serial ( pipeline , input_iterator , output_dir , output_file_base = None ) : if isinstance ( pipeline . output_type , dict ) : for name , type_ in pipeline . output_type . items ( ) : if not hasattr ( type_ , 'SerializeToString' ) : raise ValueError ( 'Pipeline output "%s" does not have method SerializeToString. ' 'Output type = %s' % ( name , pipeline . output_type ) ) else : if not hasattr ( pipeline . output_type , 'SerializeToString' ) : raise ValueError ( 'Pipeline output type %s does not have method SerializeToString.' % pipeline . output_type ) if not tf . gfile . Exists ( output_dir ) : tf . gfile . MakeDirs ( output_dir ) output_names = pipeline . output_type_as_dict . keys ( ) if output_file_base is None : output_paths = [ os . path . join ( output_dir , name + '.tfrecord' ) for name in output_names ] else : output_paths = [ os . path . join ( output_dir , '%s_%s.tfrecord' % ( output_file_base , name ) ) for name in output_names ] writers = dict ( ( name , tf . python_io . TFRecordWriter ( path ) ) for name , path in zip ( output_names , output_paths ) ) total_inputs = 0 total_outputs = 0 stats = [ ] for input_ in input_iterator : total_inputs += 1 for name , outputs in _guarantee_dict ( pipeline . transform ( input_ ) , list ( output_names ) [ 0 ] ) . items ( ) : for output in outputs : # pylint:disable=not-an-iterable writers [ name ] . write ( output . SerializeToString ( ) ) total_outputs += len ( outputs ) stats = statistics . merge_statistics ( stats + pipeline . get_stats ( ) ) if total_inputs % 500 == 0 : tf . logging . info ( 'Processed %d inputs so far. Produced %d outputs.' , total_inputs , total_outputs ) statistics . log_statistics_list ( stats , tf . logging . info ) tf . logging . info ( '\n\nCompleted.\n' ) tf . logging . info ( 'Processed %d inputs total. Produced %d outputs.' , total_inputs , total_outputs ) statistics . log_statistics_list ( stats , tf . logging . info )