TensorFlow: импорт данных - базовая механика, итераторы

Импортирование данных

tf.data API позволяет создавать комплексные пайплайны ввода из простых, переиспользуемых частей. Например, пайплайн для модели изображений может собирать данные из файлов в распределенной файловой системе, применять преобразования к каждому изображению, и соединять взятые в случайном порядке изображения в пакеты для тренировки. Пайплайн для текстовой модели может включать извлечение символов из сырых текстовых данных, их преобразование для встраивания идентификаторов в поисковую таблицу, и пакетизация последовательностей различной длины. tf.data API позволяет легко обрабатывать большие количества данных, различные форматы данных, и выполнять сложные трансформации.

tf.data API вводит две новые абстракции в TensorFlow:

• tf.data.Dataset представляет последовательность элементов, в которой каждый элемент содержит один или более Tensor объектов. Например, в пайплайне изображений элемент может быть единичным тренировочным примером с парой тензоров, представляющих данные изображения и метку. Существует два разных способа создать набор данных (dataset):
  • Создание источника (Dataset.from_tensor_slices()) конструирует набор данных (dataset) из одного или более tf.Tensor объектов.
  • Применение трансформации (Dataset.batch()) конструирует набор данных (dataset) из одного или более tf.data.Dataset объектов.
• tf.data.Iterator предоставляет основной способ извлечения элементов из набора данных (dataset). Операция, возвращаемая Iterator.get_next() выводит следующий элемент Dataset, когда исполняется, и обычно действует как интерфейс между пайпланом ввода и моделью. Простейший итератор - это "одноразовый итератор", который связан с отдельным Dataset и итерирует по нему один раз. Для более совершенного использования, Iterator.initializer операция позволяет реинициализировать и параметризировать итератор с различными наборами данных (datasets), так что можно, например, итерировать тренировочные и валидационные данные много раз в той же программе.

Базовая механика

Эта часть описывает основы создания разных видов Dataset и Iterator объектов, и как извлекать данные из них.

Для начала пайплайна ввода необходимо определить источник. Например, чтобы сконструировать Dataset из нескольких тензоров в памяти можно использовать tf.data.Dataset.from_tensors() или tf.data.Dataset.from_tensor_slices(). С другой стороны, если вводные данные на диске в рекомендуемом TFRecord формате, можно сконструировать tf.data.TFRecordDataset.

Как только получили Dataset объект, можно трансформировать его в новый Dataset, используя цепь вызовов методов на tf.data.Dataset объекте. Например, можно применить по-элементные трансформации, такие как Dataset.map() (для применения функции к каждому элементу), и много-элементные трансформации, такие как Dataset.batch().

Наиболее общий путь для получения данных от Dataset - это сделать объект-итератор, который предоставляет доступ к одному элемену набора (dataset'а) за один раз (например, вызвав Dataset.make_one_shot_iterator()). tf.data.Iterator предоставляет две операции: Iterator.initializer, который позволяет (ре)инициализировать состояние итератора; и Iterator.get_next(), который возвращает tf.Tensor объекты, которые соответствуют символически следующему элементу.

Структура Dataset

Dataset включает в себя элементы, каждый из которых имеет одинаковую структуру. Элемент содержит один или более tf.Tensor объектов, называемых компонентами. Каждый компонент имеет tf.DType, представляющий тип элементов в тензоре, и tf.TensorShape представляющий (возможно частично определенную) статическую форму каждого элемента. Dataset.output_types и Dataset.output_shapes свойства позволяют проверять типы и формы каждого компонента элемента набора данных (dataset). Вложенная структура этих свойств соотносится со структурой элемента, который может быть единичным тензором, таплом (tuple - тип структуры данных в python) тензоров, или вложенным таплом тензоров. Например:

dataset1 = tf.data.Dataset.from_tensor_slices(tf.random_uniform([4, 10]))
print(dataset1.output_types)  # ==> "tf.float32"
print(dataset1.output_shapes)  # ==> "(10,)"

dataset2 = tf.data.Dataset.from_tensor_slices(
   (tf.random_uniform([4]),
    tf.random_uniform([4, 100], maxval=100, dtype=tf.int32)))
print(dataset2.output_types)  # ==> "(tf.float32, tf.int32)"
print(dataset2.output_shapes)  # ==> "((), (100,))"

dataset3 = tf.data.Dataset.zip((dataset1, dataset2))
print(dataset3.output_types)  # ==> (tf.float32, (tf.float32, tf.int32))
print(dataset3.output_shapes)  # ==> "(10, ((), (100,)))"

Часто удобно давать имена каждому компоненту элемента, например если они представляют разные свойства тренировочного примера. Вдобавок к таплам (tuples) можно использовать collections.namedtuple или словарь, картирующий строки к тензорам для представления одного элемента в Dataset.

dataset = tf.data.Dataset.from_tensor_slices(
   {"a": tf.random_uniform([4]),
    "b": tf.random_uniform([4, 100], maxval=100, dtype=tf.int32)})
print(dataset.output_types)  # ==> "{'a': tf.float32, 'b': tf.int32}"
print(dataset.output_shapes)  # ==> "{'a': (), 'b': (100,)}"

Dataset трансформации поддерживают наборы данных (datasets) любой структуры. При использовании Dataset.map(), Dataset.flat_map(), и Dataset.filter() трансформаций, которые применяют функцию к каждому элементу, структура элементов определяется аргументами функции:

dataset1 = dataset1.map(lambda x: ...)

dataset2 = dataset2.flat_map(lambda x, y: ...)

dataset3 = dataset3.filter(lambda x, (y, z): ...)

Создание итератора

Как только построен Dataset для представления вводных данных, следующий шаг - создать Iterator для доступа к элементам из этого набора данных (dataset). tf.data API на данный момент поддерживает следующие итераторы, в порядке увеличения сложности:

• одноразовый (one-shot)
• инициализируемый (initializable)
• реинициализируемый (reinitializable)
• загрузочный (feedable)

Одноразовый (one-shot) итератор - это простейшая форма итератора, которая поддерживает только однократное итерирование по набору данных (dataset), без необходимости явной инициализации. Одноразовые итераторы обрабатывают почти все случаи, которые существуют в поддерживаемых пайплайнах ввода, основанных на очереди. Но они не поддерживают параметризацию. Пример использования Dataset.range():

dataset = tf.data.Dataset.range(100)
iterator = dataset.make_one_shot_iterator()
next_element = iterator.get_next()

for i in range(100):
  value = sess.run(next_element)
  assert i == value

Инициализируемый (initializable) итератор требует выполнения явной iterator.initializer операции перед его использованием. В обмен на такое неудобство это позволяет параметризировать определение набора данных (dataset), используя один или более tf.placeholder() тензоров, которые могут быть переданы при инициализации итератора. Продолжая Dataset.range() пример:

max_value = tf.placeholder(tf.int64, shape=[])
dataset = tf.data.Dataset.range(max_value)
iterator = dataset.make_initializable_iterator()
next_element = iterator.get_next()

# Инициализируем итератор для набора данных с 10 элементами.
sess.run(iterator.initializer, feed_dict={max_value: 10})
for i in range(10):
  value = sess.run(next_element)
  assert i == value

# Инициализируем тот же итератор для набора данных со 100 элементами.
sess.run(iterator.initializer, feed_dict={max_value: 100})
for i in range(100):
  value = sess.run(next_element)
  assert i == value

Реинициализируемый (reinitializable) итератор может быть инициализирован из нескольких различных Dataset объектов. Например, может существовать тренировочный пайплайн ввода, который использует преобразования изображений ввода, применяемые в случайном порядке, для улучшения генерализации, и валидационный пайплайн ввода, который оценивает прогнозы на немодифицированных данных. Эти пайплайны обычно используют разные Dataset объекты с одной и той же структурой (то есть одинаковые типы и совместимые формы для каждого компонента).

# Определяем тренировочный и валидационный наборы данных 
# с одинаковой структурой.
training_dataset = tf.data.Dataset.range(100).map(
    lambda x: x + tf.random_uniform([], -10, 10, tf.int64))
validation_dataset = tf.data.Dataset.range(50)

# Реинициализируемый итератор определяется этими структурами.
# Можно использовать output_types и output_shapes свойства
# объектов training_dataset или validation_dataset здесь,
# поскольку они совместимы.
iterator = tf.data.Iterator.from_structure(
                                     training_dataset.output_types,
                                     training_dataset.output_shapes)
next_element = iterator.get_next()

training_init_op = iterator.make_initializer(training_dataset)
validation_init_op = iterator.make_initializer(validation_dataset)

# Выполняем 20 эпох, в которых тренировочный набор данных итерируется,
# а следом и валидационный набор.
for _ in range(20):
  # Инициализируем итератор для тренировочного набора.
  sess.run(training_init_op)
  for _ in range(100):
    sess.run(next_element)

  # Инициализируем итератор для валидационного набора.
  sess.run(validation_init_op)
  for _ in range(50):
    sess.run(next_element)

Загружаемый (feedable) итератор может быть использован вместе с tf.placeholder для выбора того какой Iterator использовать в каждом вызове tf.Session.run посредством feed_dict механизма. Он предоставляет ту же функциональность, что и реинициализируемый итератор, но он не требует инициализации итератора со старта набора данных, когда переключаемся между итераторами. Например, используя те же тренировочный и валидационный пример как выше, можно использовать tf.data.Iterator.from_string_handle, чтобы определить загружаемый итератор, который позволяет переключаться между двумя наборами данных:

# Определяем тренировочный и валидационный датасеты 
# с одинаковой структурой
training_dataset = tf.data.Dataset.range(100).map(
    lambda x: x + tf.random_uniform([], -10, 10, tf.int64)).repeat()
validation_dataset = tf.data.Dataset.range(50)

# Загружаемый итератор определен обработчиком placeholder
# и его структурой.
# Можно использовать output_types и output_shapes свойства
# объектов training_dataset или validation_dataset здесь,
# поскольку они имеют одинаковую структуру.
handle = tf.placeholder(tf.string, shape=[])
iterator = tf.data.Iterator.from_string_handle(
    handle, training_dataset.output_types, 
            training_dataset.output_shapes)
next_element = iterator.get_next()

# Можно использовать загружаемый итератор 
# с различными другими итераторами,
# такими как одноразовый и инициализируемый
training_iterator = training_dataset.make_one_shot_iterator()
validation_iterator = validation_dataset.make_initializable_iterator()

# Метод `Iterator.string_handle()` возвращает тензор, 
# который может быть оценен
# и использован для наполнения `handle` placeholder.
training_handle = sess.run(training_iterator.string_handle())
validation_handle = sess.run(validation_iterator.string_handle())

# Итерируем постоянно, переключаясь между тренировкой и валидацией.
while True:
  # Проходим 200 шагов, используя тренировочный датасет.
  # Отметим, что тренировочный датасет бесконечный,
  # и мы продолжаем с того места, где остановились на
  # предыдущей итерации `while` цикла.
  for _ in range(200):
    sess.run(next_element, feed_dict={handle: training_handle})

  # Делаем один проход по валидационному датасету.
  sess.run(validation_iterator.initializer)
  for _ in range(50):
    sess.run(next_element, feed_dict={handle: validation_handle})

Получение значений из итератора

Метод Iterator.get_next() возвращает один или более tf.Tensor объектов, которые соотвествуют символически следующему элементу итератора. Каждый раз эти тензоры оцениваются, они принимают значение следующего элемента в подлежащем датасете. (Отметим что, как и другие объекты состояния в TensorFlow, вызываемый Iterator.get_next() не сразу продвигает итератор. Необходимо использовать возвращаемые tf.Tensor объекты в TensorFlow выражении, и передавать результат этого выражения в tf.Session.run(), чтобы получить следующиь элементы и продвинуть итератор.)

Если итератор достигает конца датасета (набора данных), исполнение Iterator.get_next() операции вызовет tf.errors.OutOfRangeError исключение. После этой точки итератор будет в неиспользуемом состоянии, и для дальнейшего использования его необходимо инициализировать снова.

dataset = tf.data.Dataset.range(5)
iterator = dataset.make_initializable_iterator()
next_element = iterator.get_next()

# Обычно `result` будет выводом модели,
# или тренировочной операцией оптимизиатора.
result = tf.add(next_element, next_element)

sess.run(iterator.initializer)
print(sess.run(result))  # ==> "0"
print(sess.run(result))  # ==> "2"
print(sess.run(result))  # ==> "4"
print(sess.run(result))  # ==> "6"
print(sess.run(result))  # ==> "8"
try:
  sess.run(result)
except tf.errors.OutOfRangeError:
  print("End of dataset")  # ==> "End of dataset"

Общепринятный паттерн - обернуть "тренировочный цикл" в try-except блок:

sess.run(iterator.initializer)
while True:
  try:
    sess.run(result)
  except tf.errors.OutOfRangeError:
    break

Если каждый элемент датасета имеет вложенную структуру, то возвращаемое значение Iterator.get_next() будет одним или более tf.Tensor объектов в той же вложенной структуре:

dataset1 = tf.data.Dataset.from_tensor_slices(
                                       tf.random_uniform([4, 10]))
dataset2 = tf.data.Dataset.from_tensor_slices(
            (tf.random_uniform([4]), tf.random_uniform([4, 100])))
dataset3 = tf.data.Dataset.zip((dataset1, dataset2))

iterator = dataset3.make_initializable_iterator()

sess.run(iterator.initializer)
next1, (next2, next3) = iterator.get_next()

Отметим, что next1, next2, и next3 - это тензоры, произведенные одним и тем же опом(op)/узлом(node), созданным Iterator.get_next(). Поэтому оценка любого из этих тензоров продвинет итератор для всех компонентов. Обычный приемник будет включать все компоненты в одном выражении.

Сохранение состояние итератора

Функция tf.contrib.data.make_saveable_from_iterator создает SaveableObject из итератора, который может быть использован для сохранения и восстановления текущего состояния итератора (и целого пайплайна ввода). Сохраняемый объект, созданный таким образом, может быть добавлен в tf.train.Saver лист переменных или tf.GraphKeys.SAVEABLE_OBJECTS коллекцию для сохранения и восстановления в той же манере как tf.Variable.

# Создаем сохраняемый объект из итератора.
saveable = tf.contrib.data.make_saveable_from_iterator(iterator)

# Сохраняем состояние итератора добавлением его 
# к коллекции сохраняемых объектов.
tf.add_to_collection(tf.GraphKeys.SAVEABLE_OBJECTS, saveable)
saver = tf.train.Saver()

with tf.Session() as sess:

  if should_checkpoint:
    saver.save(path_to_checkpoint)

# Восстанавливаем состояние итератора.
with tf.Session() as sess:
  saver.restore(sess, path_to_checkpoint)

О дальнейших возможностях импорта данных в TensorFlow читайте в следующих постах.