Как улучшить производительность этого конвейера данных для моей модели tensorflow - программирование
Подтвердить что ты не робот

Как улучшить производительность этого конвейера данных для моей модели tensorflow

У меня есть модель тензорного потока, которую я тренирую на google-colab. Фактическая модель более сложная, но я сконденсировал ее в воспроизводимый пример (удаляли сохранение/восстановление, распад скорости обучения, утверждает, события тензорной группы, отсечение градиента и т.д.). Модель работает разумно (сходится к приемлемой потере), и я ищу способ ускорить обучение (итерации в секунду).

В настоящее время на GPU colab требуется 10 минут для тренировки на 1000 итераций. С моим текущим размером партии 512 это означает, что модель обрабатывает ~ 850 примеров в секунду (я бы предпочел иметь размер партии 512, если другие размеры не обеспечивают разумного ускорения. Сам по себе изменение размера партии не изменяет скорость).

Поэтому в настоящее время у меня есть данные, хранящиеся в формате tfrecord: вот пример файла 500 Мб, общий размер данных - ~ 0.5 Тб. Эти данные проходят через достаточно тяжелый шаг предварительной обработки (я не могу предварительно выполнить предварительную обработку, так как это увеличит размер моих tfrecords намного выше того, что я могу себе позволить). Предварительная обработка выполняется через tf.data, а выходные тензоры ((batch_size, 8, 8, 24) которые обрабатываются как NHWC (batch_size, 10)), передаются в модель. Пример colab не содержит упрощенной модели, которая служит в качестве примера.

Я попробовал несколько подходов к ускорению обучения:

  • ручное размещение устройства (данные предварительная обработка на процессоре, propagations на ГПЕ), но все мои попытки привели к ухудшению скорости (от 10% до 50% прироста).
  • улучшить предварительную обработку данных. Я рассмотрел учебники по видео и данным tf.data. Я пробовал почти все техники из этого учебника, не получал улучшения (снижение скорости от 0% до 15%). В частности, я попытался:
    • dataset.prefetch(...)
    • прохождение num_parallel_calls к карте
    • объединение карты и партии в tf.contrib.data.map_and_batch
    • использование parallel_interleave

Здесь приведен код, связанный с предварительной обработкой данных (вот полный воспроизводимый пример с примерами данных):

_keys_to_map = {
    'd': tf.FixedLenFeature([], tf.string),  # data
    's': tf.FixedLenFeature([], tf.int64),   # score
}


def _parser(record):][3]
    parsed = tf.parse_single_example(record, _keys_to_map)
    return parsed['d'], parsed['s']


def init_tfrecord_dataset():
  files_train = glob.glob(DIR_TFRECORDS + '*.tfrecord')
  random.shuffle(files_train)

  with tf.name_scope('tfr_iterator'):
    ds = tf.data.TFRecordDataset(files_train)      # define data from randomly ordered files
    ds = ds.shuffle(buffer_size=10000)             # select elements randomly from the buffer
    ds = ds.map(_parser)                           # map them based on tfrecord format
    ds = ds.batch(BATCH_SIZE, drop_remainder=True) # group elements in batch (remove batch of less than BATCH_SIZE)
    ds = ds.repeat()                               # iterate infinitely 

    return ds.make_initializable_iterator()        # initialize the iterator


def iterator_to_data(iterator):
  """Creates a part of the graph which reads the raw data from an iterator and transforms it to a 
  data ready to be passed to model.

  Args:
    iterator      - iterator. Created by 'init_tfrecord_dataset'

  Returns:
    data_board      - (BATCH_SIZE, 8, 8, 24) you can think about as NWHC for images.
    data_flags      - (BATCH_SIZE, 10)
    combined_score  - (BATCH_SIZE,)
  """

  b = tf.constant((128, 64, 32, 16, 8, 4, 2, 1), dtype=tf.uint8, name='unpacked_const')

  with tf.name_scope('tfr_parse'):
    with tf.name_scope('packed_data'):
      next_element = iterator.get_next()
      data_packed, score_int = next_element
      score = tf.cast(score_int, tf.float64, name='score_float')

    # https://stackoverflow.com/q/45454470/1090562
    with tf.name_scope('data_unpacked'):
      data_unpacked = tf.reshape(tf.mod(tf.to_int32(tf.decode_raw(data_packed, tf.uint8)[:,:,None] // b), 2), [BATCH_SIZE, 1552], name='data_unpack')

    with tf.name_scope('score'):
      with tf.name_scope('is_mate'):
        score_is_mate = tf.cast(tf.squeeze(tf.slice(data_unpacked, [0, 1546], [BATCH_SIZE, 1])), tf.float64, name='is_mate')
      with tf.name_scope('combined'):
        combined_score = (1 - score_is_mate) * VALUE_A * tf.tanh(score / VALUE_K) + score_is_mate * tf.sign(score) * (VALUE_A + (1 - VALUE_A) / (VALUE_B - 1) * tf.reduce_max(tf.stack([tf.zeros(BATCH_SIZE, dtype=tf.float64), VALUE_B - tf.abs(score)]), axis=0))


    with tf.name_scope('board'):
      with tf.name_scope('reshape_layers'):
        data_board = tf.reshape(tf.slice(data_unpacked, [0, 0], [BATCH_SIZE, 8 * 8 * 24]), [BATCH_SIZE, 8, 8, 24], name='board_reshape')

      with tf.name_scope('combine_layers'):  
        data_board = tf.cast(tf.stack([
          data_board[:,:,:, 0],
          data_board[:,:,:, 4],
          data_board[:,:,:, 8],
          data_board[:,:,:,12],
          data_board[:,:,:,16],
          data_board[:,:,:,20],
          - data_board[:,:,:, 1],
          - data_board[:,:,:, 5],
          - data_board[:,:,:, 9],
          - data_board[:,:,:,13],
          - data_board[:,:,:,17],
          - data_board[:,:,:,21],
          data_board[:,:,:, 2],
          data_board[:,:,:, 6],
          data_board[:,:,:,10],
          data_board[:,:,:,14],
          data_board[:,:,:,18],
          data_board[:,:,:,22],
          - data_board[:,:,:, 3],
          - data_board[:,:,:, 7],
          - data_board[:,:,:,11],
          - data_board[:,:,:,15],
          - data_board[:,:,:,19],
          - data_board[:,:,:,23],
        ], axis=3), tf.float64, name='board_compact')

    with tf.name_scope('flags'):
      data_flags = tf.cast(tf.slice(data_unpacked, [0, 1536], [BATCH_SIZE, 10]), tf.float64, name='flags')

  return data_board, data_flags, combined_score

Я ищу практические решения (я пробовал много теоретических идей), которые могут улучшить скорость обучения (с точки зрения примеров/вторых). Я не ищу способ улучшить точность модели (или изменить модель), поскольку это всего лишь тестовая модель.

Я потратил значительное количество времени, пытаясь оптимизировать это (и сдался). Поэтому я был бы рад наградить 200 долларов за рабочее решение с хорошим объяснением.

4b9b3361

ОТВЕТЫ

Ответ 1

Предложение от hampi для профилирования вашей учебной работы является хорошим, и может быть необходимо, чтобы понять фактические узкие места в вашем трубопроводе. Другие рекомендации в руководстве по производительности Input Pipeline также должны быть полезны.

Тем не менее, есть другое возможное "быстрое решение", которое может быть полезным. В некоторых случаях объем работы в преобразовании Dataset.map() может быть очень небольшим, и в нем преобладает стоимость вызова функции для каждого элемента. В этих случаях мы часто пытаемся векторизовать функцию карты и перемещать ее после преобразования Dataset.batch(), чтобы вызывать функцию меньше раз (в данном случае 1/512 раз) и выполнять больше: и потенциально легче распараллелить - операции над каждым пакетом. К счастью, ваш конвейер может быть векторизован следующим образом:

def _batch_parser(record_batch):
  # NOTE: Use 'tf.parse_example()' to operate on batches of records.
  parsed = tf.parse_example(record_batch, _keys_to_map)
  return parsed['d'], parsed['s']

def init_tfrecord_dataset():
  files_train = glob.glob(DIR_TFRECORDS + '*.tfrecord')
  random.shuffle(files_train)

  with tf.name_scope('tfr_iterator'):
    ds = tf.data.TFRecordDataset(files_train)      # define data from randomly ordered files
    ds = ds.shuffle(buffer_size=10000)             # select elements randomly from the buffer
    # NOTE: Change begins here.
    ds = ds.batch(BATCH_SIZE, drop_remainder=True) # group elements in batch (remove batch of less than BATCH_SIZE)
    ds = ds.map(_batch_parser)                     # map batches based on tfrecord format
    # NOTE: Change ends here.
    ds = ds.repeat()                               # iterate infinitely 

    return ds.make_initializable_iterator()        # initialize the iterator

В настоящее время векторизация - это изменение, которое вы должны внести вручную, но команда tf.data работает над tf.data оптимизации, который обеспечивает автоматическую векторизацию.

Ответ 2

У меня есть несколько предложений:

1) После создания партии вся партия обрабатывается функцией iterator_to_data(). На самом деле это не распространяется на несколько потоков, по крайней мере, не на уровне api. Вместо этого вы можете попробовать что-то подобное в функции init_tfrecord_dataset():

ds = tf.data.TFRecordDataset(files_train)      # define data from randomly ordered files
ds = ds.shuffle(buffer_size=10000)             # select elements randomly from the buffer
ds = ds.map(_parser)  
ds = ds.map(map_func=iterator_to_data, num_parallel_calls=FLAGS.num_preprocessing_threads)
ds = ds.batch(BATCH_SIZE, drop_remainder=True) # group elements in batch (remove batch of less than BATCH_SIZE)
ds = ds.repeat()

вы также можете изменить несколько строк в функции iterator_to_data(), поскольку входной аргумент не является итератором с указанными выше изменениями.

2) Вы также можете получить информацию о профилировании, используя что-то вроде tf.train.ProfilerHook. Это может сказать вам, есть ли узкое место с процессором или gpu. Например, если узкое место с процессором, вы можете увидеть, что GPU ops ждет завершения операции memcpyHtoD.

Ответ 3

Если вы можете обеспечить захват TPU/GPU, как упомянуто в видео производительности ниже, это поможет в выявлении узкого места и лучшего предложения. Кроме того, колаб, которым вы поделились, больше не доступен :(, поэтому вы не можете видеть код или данные.

https://www.youtube.com/watch?v=SxOsJPaxHME