【技术分享】TFRecord 实践

An Example for a movie recommendation application:
   features {
     feature {
       key: "age"
       value { float_list {
         value: 29.0
       }}
     }
     feature {
       key: "movie"
       value { bytes_list {
         value: "The Shawshank Redemption"
         value: "Fight Club"
       }}
     }
     feature {
       key: "movie_ratings"
       value { float_list {
         value: 9.0
         value: 9.7
       }}
     }
     feature {
       key: "suggestion"
       value { bytes_list {
         value: "Inception"
       }}
     }

context: {
   feature: {
     key  : "locale"
     value: {
       bytes_list: {
         value: [ "pt_BR" ]
       }
     }
   }
   feature: {
     key  : "age"
     value: {
       float_list: {
         value: [ 19.0 ]
       }
     }
   }
   feature: {
     key  : "favorites"
     value: {
       bytes_list: {
         value: [ "Majesty Rose", "Savannah Outen", "One Direction" ]
       }
     }
   }
 }
 feature_lists: {
   feature_list: {
     key  : "movie_ratings"
     value: {
       feature: {
         float_list: {
           value: [ 4.5 ]
         }
       }
       feature: {
         float_list: {
           value: [ 5.0 ]
         }
       }
     }
   }
   feature_list: {
     key  : "movie_names"
     value: {
       feature: {
         bytes_list: {
           value: [ "The Shawshank Redemption" ]
         }
       }
       feature: {
         bytes_list: {
           value: [ "Fight Club" ]
         }
       }
     }
   }
   feature_list: {
     key  : "actors"
     value: {
       feature: {
         bytes_list: {
           value: [ "Tim Robbins", "Morgan Freeman" ]
         }
       }
       feature: {
         bytes_list: {
           value: [ "Brad Pitt", "Edward Norton", "Helena Bonham Carter" ]
         }
       }
     }
   }
 }

### other function

def _int64_feature(value):
  """Wrapper for inserting an int64 Feature into a SequenceExample proto."""
  return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))


def _bytes_feature(value):
  """Wrapper for inserting a bytes Feature into a SequenceExample proto."""
  return tf.train.Feature(bytes_list=tf.train.BytesList(value=[str(value)]))


def _int64_feature_list(values):
  """Wrapper for inserting an int64 FeatureList into a SequenceExample proto."""
  return tf.train.FeatureList(feature=[_int64_feature(v) for v in values])


def _bytes_feature_list(values):
  """Wrapper for inserting a bytes FeatureList into a SequenceExample proto."""
  return tf.train.FeatureList(feature=[_bytes_feature(v) for v in values])

### 
with tf.gfile.FastGFile(image.filename, "r") as f:
  encoded_image = f.read()

try:
  decoder.decode_jpeg(encoded_image) ## tensor->nparray
except (tf.errors.InvalidArgumentError, AssertionError):
  print("Skipping file with invalid JPEG data: %s" % image.filename)
  return

context = tf.train.Features(feature={
    "image/image_id": _int64_feature(image.image_id),
    "image/data": _bytes_feature(encoded_image),
}) ## context config

assert len(image.captions) == 1
caption = image.captions[0]
caption_ids = [vocab.word_to_id(word) for word in caption]
feature_lists = tf.train.FeatureLists(feature_list={
    "image/caption": _bytes_feature_list(caption),
    "image/caption_ids": _int64_feature_list(caption_ids)
}) ## feature list config
sequence_example = tf.train.SequenceExample(
    context=context, feature_lists=feature_lists) ## sequence Example

# serialized data
 serialized = [
    features
      { feature { key: "ft" value { float_list { value: [1.0, 2.0] } } } },
    features
      { feature []},
    features
      { feature { key: "ft" value { float_list { value: [3.0] } } }
  ]
# VarlenFeature的使用方法是
features={
    "ft":tf.VarLenFeature(tf.float32)
}
# get
{"ft": SparseTensor(indices=[[0, 0], [0, 1], [2, 0]],
                      values=[1.0, 2.0, 3.0],
                      dense_shape=(3, 2)) }

import tensorflow as tf
import os
keys=[[1.0,2.0],[2.0,3.0]]
sess=tf.InteractiveSession()
sess.run(tf.global_variables_initializer())
def make_example(locale,age,score,times):
    example = tf.train.SequenceExample(
        context=tf.train.Features(
            feature={    "locale":tf.train.Feature(bytes_list=tf.train.BytesList(value=[locale])),       "age":tf.train.Feature(int64_list=tf.train.Int64List(value=[age]))
        }),
        feature_lists=tf.train.FeatureLists(
            feature_list={
 "movie_rating":tf.train.FeatureList(feature=[tf.train.Feature(float_list=tf.train.FloatList(value=score)) for i in range(times)])
            }
        )
    )
    return example.SerializeToString()
context_features = {
    "locale": tf.FixedLenFeature([],dtype=tf.string),
    "age": tf.FixedLenFeature([],dtype=tf.int64)
}
sequence_features = {
    "movie_rating": tf.FixedLenSequenceFeature([3], dtype=tf.float32,allow_missing=True)
}
context_parsed, sequence_parsed  = tf.parse_single_sequence_example(make_example("china",24,[1.0,3.5,4.0],2),context_features=context_features,sequence_features=sequence_features)
print tf.contrib.learn.run_n(context_parsed)
print tf.contrib.learn.run_n(sequence_parsed)
# get
[{'locale': 'china', 'age': 24}]

[{'movie_rating': array([[ 1. ,  3.5,  4. ],
       [ 1. ,  3.5,  4. ]], dtype=float32)}]

example = ...ops to create one example...
# Create a queue, and an op that enqueues examples one at a time in the queue.
queue = tf.RandomShuffleQueue(...)
enqueue_op = queue.enqueue(example)
# Create a training graph that starts by dequeuing a batch of examples.
inputs = queue.dequeue_many(batch_size)
train_op = ...use 'inputs' to build the training part of the graph...

writer = tf.python_io.TFRecordWriter(filename)
writer.write(serilized_object)
writer.close()

Batching

def read_my_file_format(filename_queue):
 reader = tf.SomeReader() # tf.TFRecordReader()
 key, record_string = reader.read(filename_queue) # queue
 example, label = some_decoder(record_string) # parse example opt
 processed_example = some_processing(example) # example processing
 return processed_example, label

def input_pipeline(filenames, batch_size, num_epochs=None):
# put TFRecord into queue, num_epochs is number of epochs and  need to be local_initialized, if not specified, it will consistent read until the queue is outOfRange.
 filename_queue = tf.train.string_input_producer(
     filenames, num_epochs=num_epochs, shuffle=True) 
 example, label = read_my_file_format(filename_queue)
 # min_after_dequeue defines how big a buffer we will randomly sample
 #   from -- bigger means better shuffling but slower start up and more
 #   memory used.
 # capacity must be larger than min_after_dequeue and the amount larger
 #   determines the maximum we will prefetch.  Recommendation:
 #   min_after_dequeue + (num_threads + a small safety margin) * batch_size
 min_after_dequeue = 10000
 capacity = min_after_dequeue + 3 * batch_size
 example_batch, label_batch = tf.train.shuffle_batch(
     [example, label], batch_size=batch_size, capacity=capacity,
     min_after_dequeue=min_after_dequeue)
 return example_batch, label_batch

import tensorflow as tf

def run_training():
    with tf.Graph().as_default(), tf.Session() as sess:
        datas,labels = input_pipeline("example.tfrecords",32)
        c = config()
        initializer = tf.random_uniform_initializer(-1*c.init_scale,1*c.init_scale)
        with tf.variable_scope("model",initializer=initializer):
            model = ModelFunction(config=c,data=datas,label=labels)

        fetches = [model.train_op,model.accuracy,model.loss]

        #init
        init_op = tf.group(tf.global_variables_initializer(),
                       tf.local_variables_initializer())
        sess.run(init_op)

        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess=sess,coord=coord)
        try:
            while not coord.should_stop():
                _,accuracy,loss= sess.run(fetches)
        except tf.errors.OutOfRangeError:
            print("done training")
        finally:
            coord.request_stop()
        coord.join(threads)
        sess.close()

def main():
    run_training()

if __name__=='__main__':
    main()