[Keras, TF2.0] 딥러닝 모델에서 효율적인 입력 파이프라인 만들기

tf.data API로 성능 향상하기

• Licensed under the Apache License, Version 2.0 (the "License")
• MIT License
• https://www.tensorflow.org/guide/data_performance#%EC%B6%94%EC%83%81%EC%A0%81_%EC%A0%91%EA%B7%BC

1. 개요

• GPU와 TPU는 하나의 학습 단계를 실행하는데 필요한 시간을 급격하게 줄일 수 있다
• 최대 성능을 위해서는 현재 단계가 종료 되기 전에 다음 스텝의 데이터를 운반하는 효율적인 입력 파이프라인이 필요하다
  • tf.data API는 유연하고 효율적인 입력 파이프라인을 만드는데 도움이 된다
• 이 문서는 고성능 텐서플로 입력 파이프라인을 만드는 방법과 tf.data API의 특징을 설명한다
• tf.data API 사용법을 익히려면 텐서플로 입력 파이프라인 빌드하기 가이드를 읽는다

2. 참고자료

• 텐서플로 입력 파이프라인 빌드하기
• tf.data.Dataset API

3. 설정

try:
    %tensorflow_version 2.x
except Exception:
    pass

import tensorflow as tf

import time

• 데이터셋을 반복하고 성능을 측정한다

• 재현가능한 벤치마크를 만드는 것은 다음과 같은 요인들로 인해 어려울 수 있다

  • 현재 CPU로드
  • 네트워크 트래픽
  • 캐시와 같은 복잡한 메커니즘 등이 있다

• 따라서 재현 가능한 벤치마크를 제공하기 위해 인공 예제를 빌드한다

3-1. 데이터셋

• tf.data.Dataset 에서 상속하여 ArtificialDataset이라 불리는 클래스를 정의한다
• 데이터셋의 구성
  • num_samples(기본값3)개의 샘플을 생성
  • 첫 번째 항목이 파일 열기를 시뮬레이션하기 전에 일정 시간 sleep
  • 파일에서 데이터 읽기를 시뮬레이션하기 위해 각 항목을 생성하기 전 일정 시간 sleep

class ArtificialDataset(tf.data.Dataset):
  def _generator(num_samples):
    # 파일 열기
    time.sleep(0.03)

    for sample_idx in range(num_samples):
      # 파일에서 데이터 읽기
      time.sleep(0.015)

      yield(sample_idx, )

  def __new__(cls, num_samples=3):
    return tf.data.Dataset.from_generator(
        cls._generator,
        output_types=tf.dtypes.int64,
        output_shapes=(1,),
        args=(num_samples,)
    )

• 이 데이터셋은 tf.data.Dataset.range와 유사하며 각 샘플의 시작과 사이에 일정한 지연시간을 추가한다

3-2. 훈련 루프

• 데이터셋을 반복하는 데 걸리는 시간을 측정하는 더미 훈련 루프를 작성한다, 훈련 시간이 시뮬레이션 된다

def benchmark(dataset, num_epochs=2):
  start_time = time.perf_counter()
  for epoch_num in range(num_epochs):
    for sample in dataset:
      # 훈련 스텝마다 실행
      time.sleep(0.01)
  tf.print("실행 시간:", time.perf_counter() - start_time)

4. 성능 최적화

• 성능을 최적화하는 방법을 보여주기 위해 ArtificialDataset의 성능을 향상시킨다

4-1. 추상적 접근

• 트릭 없이 추상적 파이프라인으로 시작하여 데이터셋을 그대로 반복한다

benchmark(ArtificialDataset())

실행 시간: 0.23475170100005016

• 실제로는 다음과 같이 실행 시간이 소비됩니다

• 다음을 포함한 훈련 스텝을 수행합니다

  • 파일 열기
  • 파일에서 데이터 항목 가져오기
  • 훈련할 데이터 사용하기

• 파이프라인이 데이터를 가져 오는 동안 모델이 유휴 상태이다

• 모델이 훈련하는 동안 입력 파이프라인이 유휴 상태이다

• 따라서 훈련 스텝의 시간은 열기, 읽기, 훈련 시간의 합계이다

4-2. 가져오기(Prefetching)

• 가져오기는 전처리와 훈련 스텝의 모델 실행을 오버랩한다
• 모델이 s스텝 훈련을 실행하는 동안 입력 파이프라인은 s+1스텝의 데이터를 읽는다
  • 이렇게 하면 훈련을 하는 최대 스텝 시간과 데이터를 추출하는 데 걸리는 시간을 단축시킬 수 있다

tf.data.Dataset.prefetch - 소프트웨어 파이프라이닝 방법 제공

• 데이터가 소비되는 시간과 데이터가 생성되는 시간 간의 의존성을 줄인다
• 백그라운드 스레드와 내부 버퍼를 사용하여 요청된 시간 전에 입력 데이터셋에서 데이터를 가져온다
• 가져올 데이터의 수는 하나의 훈련 스텝에서 소비한 배치의 수와 같거나 커야한다**
• 값을 수동으로 조정하거나 **tf.data.experimental.AUTOTUNE으로 설정하면 tf.data 런타임이 실행 시에 동적으로 값을 조정한다

benchmark(
    ArtificialDataset()
    .prefetch(tf.data.experimental.AUTOTUNE)
)

실행 시간: 0.1904322709999633

4-3. 데이터 추출 병렬화

• 입력 데이터를 모든 컴퓨터에 복제하는 것은 적절하지 않기 때문에 입력 데이터를 원격으로 저장할 수 있다
• 다음과 같은 로컬과 원격 저장소의 차이 때문에 원격으로 데이터를 읽을 때 입출력에 병목이 발생할 수 있다
  • 첫 번째 바이트(Time-to-first-btye)
    • 원격 저장소에서 파일의 첫 번째 바이트를 읽는 것은 로컬 저장소에서 읽어 들이는 것보다 훨씬 오래 걸린다
  • 읽기 처리량(Read throughput)
    • 원격 저장소는 보통 큰 총 대역폭을 가지지만 하나의 파일을 읽을 때 이 대역폭의 일부만 활용할 수 있다

tf.data.Dataset.interleave - 데이터 추출 오버헤드 줄이기

• 다른 데이터셋의 내용을 인터리빙(interleaving, 끼워넣기)하여 데이터 추출 단계를 병렬화하는데 사용
• cycle_length : 중첩할 데이터셋 지정된다
• num_parallel_calls : 병렬처리 수준 지정된다
• tf.data.experimental.AUTOTUNE : 어떤 수준의 병렬처리가 tf.data런타임에 사용되는지 결정

순차적 인터리브

• tf.data.Dataset.interleave 변환의 기본 인수는 두 개의 데이터셋에서 단일 샘플을 순차적으로 인터리브한다

benchmark(
    tf.data.Dataset.range(2)
    .interleave(ArtificialDataset)
)

실행 시간: 0.1961646520001068

• Interleave 변환의 결과이다, 두 데이터셋에서 샘플을 가져온다
• 성능 향상이 되진 않는다

병렬 인터리브

• Interleave 변환의 num_parallel_calls 인수를 사용한다.
• 여러 병렬 데이터셋을 불러오고, 파일을 여는 데 기다리는 시간을 단축한다

benchmark(
    tf.data.Dataset.range(2)
    .interleave(
        ArtificialDataset,
        num_parallel_calls=tf.data.experimental.AUTOTUNE
    )
)

• 읽은 두 데이터셋이 병렬화되어 전역 데이터 처리 시간이 줄어든다

4-4. 데이터 변환 병렬화

• 데이터를 준비할 때, 입력 요소들의 전처리가 필요할 수 있다

tf.data.Dataset.map - 사용자 정의 함수를 데이터 셋의 각 요소에 적용

• 입력 요소가 서로 독립적이기 때문에 전처리는 여러 개의 CPU 코어에서 병렬로 실행될 수 있다
• num_parallel_calls : 병렬 처리 레벨을 지정한다
  • 가장 좋은 값은 hardware, data size, data shape, map function cost, cpu에서 동시에 처리되는 일에 따라 다르다
  • 단순하게 cpu 코어의 숫자로 설정할 수 있다
  • num_parallel_calls를 가용 cpu 코어 숫자보다 크게 설정한다면 비효율적인 스케줄링으로 느려질 것이다
  • tf.data.experimental.AUTOTUNE : tf.data 런타임에 가용되는 병렬화 수준 결정

def mapped_function(s):
    # Do some hard pre-processing
    tf.py_function(lambda: time.sleep(0.03), [], ())
    return s

순차적 매핑

• 병렬 처리 없이 map 변환을 기본 예제로 사용

benchmark(
    ArtificialDataset()
    .map(mapped_function)
)

실행 시간: 0.41830046699999457

• 추상적 접근의 경우 열기. 읽기. 전처리, 훈련 스텝에 소요된 시간이 합산

병렬 매핑

• 동일한 전처리 함수를 사용하지만 여러 샘플에 병렬로 적용

benchmark(
    ArtificialDataset()
    .map(
        mapped_function,
        num_parallel_calls=tf.data.experimental.AUTOTUNE
    )
)

실행 시간: 0.2524083190000965

• 그림에서 전처리 단계가 겹치므로 단일 반복의 전체 시간이 줄어든다

4-5. 캐시하기

tf.data.Dataset.cache - 데이터 셋을 메모리 또는 로컬 저장소에 캐시

• 각 에포크 동안 실행되는 일부 작업(파일 열기, 데이터 읽기 등)이 저장된다

benchmark(
    ArtificialDataset()
    .map(  # 캐시 전 시간이 많이 걸리는 작업 적용
        mapped_function
    ).cache(
    ),
    5
)

실행 시간: 0.36419766799997433

• Cache 이전의 작업(파일 열기, 데이터 읽기 등)은 첫 번째 에포크 동안에만 실행
  • 다음 에포크에는 캐시된 데이터 재사용
• Map 함수에 전달된 사용자 정의 함수가 비싸면 결과 데이터셋이 메모리 또는 로컬 스토리지에 적합할 수 있는 한 map 변환후 cache 변환을 적용한다
• 사용자 정의 함수가 캐시 용량을 넘어서 데이터셋을 저장하는 데 필요한 공간을 늘리면 cache 변환후 데이터셋을 적용하거나 훈련 작업 전에 데이터를 전처리하여 리소스 사용량을 줄인다

4-6. 매핑 벡터화

• Map 변환으로 전달된 사용자 정의 함수를 호출하면 사용자 정의 함수의 스케줄링 및 실행과 관련된 오버헤드가 있다
• 사용자 정의 함수를 벡터화(한 번에 여러 입력에 대해 작동하도록)하고 map을 변환하기 전에 배치 변환을 적용하는 것이 좋다
• 해당 사례를 설명하는데 인공 데이터셋은 적합하지 않다, 스케줄링 지연은 약 10e-6초이기 때문이다
• 이 예제에서는 tf.data.Dataset.range 함수를 사용하여 훈련 루프를 가장 단순화 한다

fast_dataset = tf.data.Dataset.range(10000)

def fast_benchmark(dataset, num_epochs=2):
    start_time = time.perf_counter()
    for _ in tf.data.Dataset.range(num_epochs):
        for _ in dataset:
            pass
    tf.print("실행 시간:", time.perf_counter() - start_time)

def increment(x):
    return x+1

스칼라 매핑

fast_benchmark(
    fast_dataset
    # 한 번에 한 항목씩 함수 적용
    .map(increment)
    # 배치
    .batch(256)
)

실행 시간: 0.87979859699999

• 위의 그림은 매핑된 함수가 각 샘플에 적용된 진행 사항을 보여준다
• 매우 빠르게 진행되지만 시간 성능에 영향을 주는 약간의 오버헤드가 있다

매핑 벡터화됨

fast_benchmark(
    fast_dataset
    .batch(256)
    # items의 배치에 함수 적용
    # tf.Tensor.__add__ 메서드는 이미 배치를 다룸
    .map(increment)
)

실행 시간: 0.027307533999987754

• 매핑된 함수가 한 번 호출되어 배치에 적용된다
• 함수를 실행하는 데 시간이 더 걸릴 수 있지만 오버헤드는 한 번만 나타나므로 전체 시간 성능이 향상된다

4-7. 메모리 사용량(footprint) 줄이기

• interleave, prefetch, shuffle을 포함한 많은 변환은 요소들의 내부 버퍼를 사용한다
• map 변환을 할 경우 경우 요소의 크기가 변경되고 map 변환의 순서와 버퍼 요소가 메모리 사용에 영향을 준다
• 순서를 다르게 하는 것이 성능에 도움이 되는 경우 메모리 사용량이 낮아지는 순서를 선택하는 것이 좋다

부분 계산 캐싱

• 변환으로 인해 데이터가 너무 커서 메모리에 맞지 않는 경우를 제외하고 map 변환 후 데이터셋을 캐시하는 것이 좋다

• 매핑된 기능을 시간 소모적인 부분과 메모리 소모적인 부분의 두 부분으로 나눌 수 있다면 교환이 성사 될 수 있다

  • 아래와 같이 변환을 연결할 수 있다

  dataset.map(time_consuming_mapping).cache().map(memory_consuming_mapping)

  • 이런 식으로 시간이 많이 걸리는 부분은 첫 번째 에포크 동안에만 실행되며 너무 많은 캐시 공간을 사용하지 않는다

5. 가장 좋은 예제 요약

• 다음은 성능이 좋은 텐서플로 입력 파이프라인을 설계하기 위한 가장 좋은 예제를 요약한 것이다
  • Prefetch 변환을 사용하여 프로듀서와 컨슈머의 작업을 오버랩한다
  • interleave 변환을 이용해 데이터 읽기 변환을 병렬화한다
  • num_parallel_calls 매개변수를 설정하여 map 변환을 병렬 처리한다
  • 데이터가 메모리에 저장될 수 있는 경우, cache 변환을 사용하여 첫 번째 에포크동안 데이터를 메모리에 캐시한다
  • map 변환에 전달된 사용자 정의 함수를 벡터화한다
  • Interleave, prefetch, shuffle 변환을 적용하여 메모리 사용을 줄인다