시계열 데이터를 다룰 때 사용하면 매우 유용합니다.
시계열 데이터를 다룰 때 다음 함수와 비슷한 것들을 직접 정의하여 sequence를 만들어주어야 하는 번거로움이 있습니다. 

def make_sequence(data, n):
    X, y = list(), list()
    
    for i in range(len(data)):
        _X = data.iloc[i:(i + n), :-1]
        if(i + n) < len(data):
            X.append(np.array(_X))
            y.append(data.iloc[i + n, -1])
        else:
            break
            
    return np.array(X), np.array(y)

tf.data를 사용하면 여러 줄로 구성되어 있는 위의 코드가 단 하나의 함수로 해결됩니다.

dataset = tf.data.Dataset.range(10)
dataset = dataset.window(5, shift=1)
for window_dataset in dataset:
  for val in window_dataset:
    print(val.numpy(), end=" ")
  print()
  • dataset.window의 첫 번째 인자는 window size이고, 두 번째는 shift 크기를 전달합니다.
  • 결과는 다음과 같습니다.

0 1 2 3 4 
1 2 3 4 5 
2 3 4 5 6 
3 4 5 6 7 
4 5 6 7 8 
5 6 7 8 9 
6 7 8 9 
7 8 9 
8 9 

결과에서 window_size = 5만큼의 데이터를 얻다가, 끝 부분에서 [6, 7, 8, 9], [7, 8, 9], ... 의 원치않는 결과를 얻고 있습니다.
이는 가져오려는 window_size가 데이터셋의 크기를 초과했기 때문에 그렇습니다.

이를 방지하기 위해 drop_remainder = True 인자를 사용합니다.

dataset = tf.data.Dataset.range(10)
dataset = dataset.window(5, shift=1, drop_remainder=True)
for window_dataset in dataset:
  for val in window_dataset:
    print(val.numpy(), end=" ")
  print()
  • for-loop를 2중으로 사용하는 이유는 dataset.windowTensor가 아닌 Dataset을 반환하기 때문입니다.
  • 이는 flat_map 함수를 사용해서 window_dataset을 flat해주어 바로 사용할 수 있습니다.
  • 이 말은 쉽게 설명하면 원래 같은 경우 5 -> 4 -> 3 -> 처럼 iter 형식으로 받을 수 있었는데, flat_map을 사용하면 [5, 4, 3, 2, 1]로 바로 받을 수 있습니다.
dataset = tf.data.Dataset.range(10)
dataset = dataset.window(5, shift=1, drop_remainder=True)
dataset = dataset.flat_map(lambda window: window.batch(5))
for window in dataset:
  print(window.numpy())

마지막으로 다음과 같이 사용할 수도 있습니다.

dataset = tf.data.Dataset.range(10)
dataset = dataset.window(5, shift=1, drop_remainder=True)
dataset = dataset.flat_map(lambda window: window.batch(5))
dataset = dataset.map(lambda window: (window[:-1], window[-1:]))
for x,y in dataset:
  print(x.numpy(), y.numpy())
  • 결과는 다음과 같습니다.
  • [0 1 2 3] [4]
    [1 2 3 4] [5]
    [2 3 4 5] [6]
    [3 4 5 6] [7]
    [4 5 6 7] [8]
    [5 6 7 8] [9]

 

reference

https://www.tensorflow.org/guide/data

 

tf.data: Build TensorFlow input pipelines  |  TensorFlow Core

The tf.data API enables you to build complex input pipelines from simple, reusable pieces. For example, the pipeline for an image model might aggregate data from files in a distributed file system, apply random perturbations to each image, and merge random

www.tensorflow.org

 

https://www.youtube.com/watch?v=n7byMbl2VUQ&list=PLQY2H8rRoyvzuJw20FG82Lgm2SZjTdIXU&index=8


머신러닝 프로세스는 크게 두 가지로 설명할 수 있습니다.

  1. 데이터 전처리
  2. 모델을 통한 연산

전처리 과정에서 우리는 CPU를 활용해서 이미지를 cropping한다던지, 기타 video 영상을 위한 처리 등을 수행합니다.
만약 전체 트레이닝 속도가 느리다면, 위 두 가지 과정 중 하나가 bottleneck일 것입니다.

GPU나 TPU는 계속해서 엄청나게 발전해왔습니다.
이들은 Matrix, linear algebra 등의 연산을 매우 빠르게 수행함으로써 ML의 속도를 향상시켰습니다.
하지만 CPU는 GPU에 비해 상대적으로 그렇게 향상되지 못했습니다.
만약 데이터 전처리 과정에서 병목 현상이 발생한다면 전체 과정이 매우 느려질 것입니다.

그래서 이번 영상에서는 tf.data를 활용하여 속도를 개선시키는 방법을 살펴볼 것입니다.
먼저, tf.data는 다들 알다시피 데이터 전처리를 위한 쉽고, 유용한 프레임워크입니다. 아래 코드에서 tf.data의 대략적인 프로세스를 볼 수 있습니다.

import tensorflow as tf

def expensive_preprocessing(record):
	pass
    
dataset = tf.data.TFRecordDataset('.../*.tfrecord')
dataset = dataset.map(expensive_preprocessing)
dataset = dataset.shuffle(buffer_size = 1024)
dataset = dataset.batch(batch_size = 128)

dataset = dataset.prefetch()

model = tf.keras.Mpdel(...)
model.fit(dataset)
  • Dataset 객체를 생성하고,
  • 일련의 preprocessing 함수를 적용하고,
  • shuffle과 batch를 결정하고,
  • prefetch() 옵션을 넣어줍니다. prefetch()는 데이터의 입력 과정에서 다음 큐에 들어갈 데이터의 전처리를 미리 병렬적으로 수행하는 함수입니다.
  • 마지막으로 모델을 학습시킵니다.

 

그렇다면 데이터 전처리의 병목 현상을 해결할 수 있는 방법은 무엇이 있을까요?

첫 번째 아이디어는 reuse computation입니다.
이 방법은 우리가 수행하는 전처리 과정에서의 연산을 한번만 사용하지 말고, 저장해두었다가 다음 연산에서도 다시 사용하는 것을 의미합니다.(약간 캐시와 비슷한 느낌?)

이 방법을 수행할 수 있도록 tf.data snapshot을 소개합니다. 데이터 전처리 과정을 저장해두었다가 사용할 수만 있다면, 모델 아키텍처를 실험하거나 여러 가지 하이퍼파라미터를 실험하는 데에 있어서 매우 유용할 것입니다.
snapshot 기능은 다음과 같이 사용할 수 있습니다.

import tensorflow as tf

def expensive_preprocessing(record):
	pass
    
dataset = tf.data.TFRecordDataset('.../*.tfrecord')
dataset = dataset.map(expensive_preprocessing)
dataset = dataset.snapshot("/path/to/snapshot_dir/") # add
dataset = dataset.shuffle(buffer_size = 1024)
dataset = dataset.batch(batch_size = 128)

dataset = dataset.prefetch()

model = tf.keras.Mpdel(...)
model.fit(dataset)
  • snapshot 기능을 사용하면 일단 한번은 전체 디스크를 활용하지만, 다음 연산부터는 이를 참고하여 연산을 수행할 수 있습니다.
  • 그리고 snapshot은 shuffle 기능을 사용하기 전에 추가해두어야 합니다. 셔플 후에 사용하면 모든 작업이 frozen되기 때문에 주의해야 합니다. 랜덤하게 입력하는 장점을 사용할 수 없죠.
    또, 이 기능은 TF 2.3부터 이용가능하다는군요.

두 번째 아이디어는 distribute computation입니다.
이 방법은 Host CPU를 만들어 worker들에게 작업을 할당한 뒤 병렬적으로 처리하고 결과는 Host CPU에서 종합하도록 합니다.
이를 tf.data serveice로 제공합니다.

사용 방법은 코드로 확인할 수 있습니다.

import tensorflow as tf

def randomized_preprocessing(record):
    pass
    
dataset = tf.data.TFRecordDataset('.../*.tfrecord')
dataset = dataset.map(randomized_preprocessing)

dataset = dataset.shuffle(buffer_size = 1024)
dataset = dataset.batch(batch_size = 32)
dataset = dataset.distribute("<master_address>") # add
dataset = dataset.prefetch()

model = tf.keras.Model(...)
model.fit(dataset)
  • distribute 이전의 코드는 cluster에서 셋팅한 worker들이 병렬적으로 수행합니다.

 

아래 코드로 내용을 대체합니다.

더 많은 층의 output을 보고 싶으면, feature_maps를 for-loop로 구현하면 됩니다.

# 신경망 시각화(조휘용)
import tensorflow as tf

get_layer_name = [layer.name for layer in model.layers]
get_output = [layer.output for layer in model.layers]

# 모델 전체에서 output을 가져올 수 있습니다.
visual_model = tf.keras.models.Model(inputs = model.input, outputs = get_output)

test_img = np.expand_dims(testX[0], axis = 0)
feature_maps = visual_model.predict(test_img)

# 첫 번째 컨볼루션 층의 특징맵을 시각화합니다.
conv_featuremap = feature_maps[0]
conv_name = get_layer_name[0]

img_size = conv_featuremap.shape[1]
img_features = conv_featuremap.shape[-1]

display_grid = np.zeros((img_size, img_size * img_features))

for i in range(img_features):
    x = conv_featuremap[:, :, :, i]
    x -= x.mean(); x /= x.std()
    x *= 64
    x += 128
    x = np.clip(x, 0, 255).astype('uint8')
    display_grid[:, i * img_size : (i + 1) * img_size] = x
    
plt.figure(figsize = (20,20))
plt.title(conv_name)
plt.grid(False)
plt.imshow(display_grid, cmap = 'viridis')

 

https://www.youtube.com/watch?v=51YtxSH-U3Y&list=PLQY2H8rRoyvzuJw20FG82Lgm2SZjTdIXU&index=7


최근 텐서플로우는 파이토치 때문에 연구에는 불편하다는 인식이 있습니다(개인적인 의견일 수도..).

이번 영상에서는 텐서플로우가 효율적인 연구를 위해 제공하는 기능을 알아보도록 하겠습니다.

 

파라미터의 상태를 제어한다는 것은 연구에서 매우 중요한 작업입니다.
예를 들어, 케라스 Dense layer의 파라미터나 bias는 층에 저장되어 있긴 하지만, 여전히 state를 다루기엔 매우 불편합니다.

더욱 편리한 제어를 위해 tf.variable_creator_scope를 사용합니다.

class FactorizedVariable(tf.Module):
    def __init__(self, a, b):
        self.a = a
        self.b = b

tf.register_tensor_conversion_function(
  FactorizedVariable, lambda x, *a, **k: tf.matmul(x.a, x.b))

def scope(next_creator, **kwargs):
    shape = kwargs['initial_value']().shape
    if len(shape) != 2: return next_creator(**kwargs)
    return FactorizedVariable(tf.Variable(tf.random.normal([shape[0], 2])),
                                         tf.Variable(tf.random.normal([2, shape[1]])))

with tf.variable_creator_scope(scope):
    d = tf.keras.layer.Dense(10)
    d(tf.zeros[20, 10])
assert isinstance(d.kernel, FactorizedVariable)
  • 먼저, 저장하고 싶은 값을 선택하고, tf.Module을 상속받은 클래스를 정의합니다.
    tf.Module은 저장하고 싶은 변수를 자동으로 추적할 수 있도록 도와줍니다.

위의 코드는 매우 간단하지만, 실제로 사용하는 모델에서는 파라미터가 매우 많기 때문에 관리가 힘듭니다. 따라서 tf.variable_creator_scope를 사용하면 자동 추적 및 파라미터의 변화를 확인할 수 있기 때문에 매우 편리합니다.

딥러닝을 연구하는 데에 있어서 계산 속도는 매우 중요합니다. 텐서플로우는 TensorFlow compiler, XLA 등을 통해 빠른 연산 속도를 지원하고 있습니다. 더욱 효과적으로 사용하려면 @tf.function(experimental_compile=True)를 사용하세요.

활성화 함수의 예를 보겠습니다. 활성화 함수에서는 element-wise 연산 때문에 속도 측면에서 부정적인 영향을 줄지도 모릅니다.
다음 예제 코드에서 속도 차이를 볼 수 있습니다.

def f(x):
    return tf.math.log(2*tf.exp(tf.nn.relu(x+1)))

c_f = tf.function(f, experimental_compile=True)
c_f(tf.zeros([100, 100]))

f = tf.function(f)
f(tf.zeros([100, 100]))

print(timeit.timeit(lambda: f(tf.zeros([100, 100])), number = 10))
# 0.007

print(timeit.timeit(lambda: c_f(tf.zeros([100, 100])), number = 10))
# 0.005 -- ~25% faster!
  • tf.function 사용은 동일합니다. 단지, experimental_compile=True를 추가합니다.
  • linear operations가 포함된 함수나 Bert를 포함한 large-scale 모델에서 효과를 볼 수 있습니다.

element-wise 연산은 옵티마이저에서도 매우 빈번하게 일어납니다. @tf.function을 옵티마이저 코드에 추가한다면 효과를 볼 수 있습니다.
다음은 직접 옵티마이저를 정의해서 @tf.function을 사용하는 예제입니다.

class MyOptimizer(tf.keras.optimizers.Optimizer):
    def __init__(self, lr, power, avg):
        super().__init__(name="MyOptimizer")
        self.lrate, self.pow, self.avg = lr, power, avg
        
    def get_config(self): pass
    def _create_slots(self, var_list):
        for v in var_list: self.add_slot(v, "accum", tf.zeros_like(v))
    
    @tf.function(experimental_compile=True)
    def _resource_apply_dense(self, grad, var, apply_state = None):
        acc = self.get_slot(var, "accum")
        acc.assign(self.avg * tf.pow(grad, self.pow) + (1-self.avg) * acc)
        
        return var.assign_sub(self.lrate * grad/tf.pow(acc, self.pow))

 

다음은 Vectorization을 이야기해보겠습니다. 이는 성능 향상을 위해 매우~! 중요한 지표입니다.
머신 러닝 모델을 다루기 위해 Vectorization이 중요하다는 것은 이미 다 알고 있는 사실이지만, 다루기가 어렵습니다.

그래서 텐서플로우는 이를 위해 auto-Vectorization을 제공합니다. 이 기능은 element-wise 연산이나 batch computation에서 성능 향상을 위해 사용될 것입니다.

Jacobian 연산을 수행하는 예제 코드입니다. jacobian은 미분값을 저장해놓은 행렬입니다.
이를 위해선 tf.GradientTape에서 tape.gradient를 무수히 호출해야하고, 다수의 for-loop를 사용하고, Tensor를 쌓아야 합니다.
이러한 과정을 거치는 코드는 언제나 작동하지만, 좀 더 효율적으로 다룰 수 있는 방법을 텐서플로우가 제공합니다.

tf.vectorized_map을 사용하는 것입니다.

x = tf.random.normal([10, 10])

with tf.GradientTape(persistent=True) as t:
    t.watch(x)
    y = tf.exp(tf.matmul(x, x))
    jac = tf.vectorized_map(
                            lambda yi: tf.vectorized_map(
                            lambda yij: t.gradient(yij, x), yi), y)
  • tf.vectorized_map을 사용하면 빠른 속도로 연산을 수행할 수 있습니다. 하지만 코드가 복잡합니다.
  • 텐서플로우는 이를 위해 jacobian을 아예 함수로 제공합니다.
x = tf.random.normal([10, 10])

with tf.GradientTape() as t:
    t.watch(x)
    y = tf.exp(tf.matmul(x, x))
jac = t.jacobian(y, x)
  • 제공하는 jacobian을 사용하면, 기존 코드보다 10배는 빠르다고 합니다.

마지막으로 데이터에 관한 이야기입니다.
텐서플로우를 사용하는 우리는 항상 매우 커다란 크기의 array를 다루게 됩니다.

 

또, 머신 러닝 모델을 다루다보면 서로 다른 타입의 데이터를 다루기도 합니다. type도 다르고, shape 다르고...
예를 들어, 텐서플로우는 다음과 같은 예를 임베딩 형태로 만들어 줍니다.

텐서플로우는 서로 다른 길이의 데이터를 다루기 위해 ragged tensor 형태를 사용합니다.

data = [['this', 'is', 'a', 'sentence'],
       ['another', 'one'],
       ['a', 'somewhat', 'longer', 'one', ',', 'this']]

rt = tf.ragged.constant(data)
vocab = tf.lookup.StaticVocabularyTable(
    tf.lookup.KeyValueTensorInitializer(
    ['This', 'is', 'a', 'sentence', 'another', 'one', 'somewhat', 'longer'],
    tf.range(8, dtype = tf.int64)), 1)

rt = tf.ragged.map_flat_values(lambda x:vocab.lookup(x), rt)
embedding_table = tf.Variable(tf.random.normal([9, 10]))
rt = tf.gather(embedding_table, rt)
tf.math.reduce_mean(rt, axis = 1)
# Result has shape (3, 10)

길이가 다르고, type이 다르면 tf.ragged를 사용하세요!

https://www.youtube.com/watch?v=v9a240kjAx4&list=PLQY2H8rRoyvzuJw20FG82Lgm2SZjTdIXU&index=4


우리는 프로젝트와 실험 등의 결과를 공유하기 위해 대표적으로 git 또는 그 외의 수단을 활용합니다.

표현하는 수많은 방법 중에서도 머신러닝을 통한 결과 공유는 차트를 활용하면 더욱 효과적으로 공유할 수 있습니다.

결과가 잘못된 경우 깃허브의 issue 기능을 통해 무엇이 잘못되었는지 토의하곤 하는데,
TensorBoard를 활용하면 결과가 왜 잘못되었는지를 명확히 알 수 있습니다.

하지만 Tensorboard를 통해 알 수 있는 결과를 스크린샷으로 공유하는 방법은 효율적이지 않습니다.
그렇기 때문에 다양한 지표를 스크린샷 형태가 아닌 다른 방법으로 표현하고 싶을지도 모릅니다.

어떻게 해야할까요?
우리는 이미 훌륭한 도구인 TensorBoard를 활용할 수 있습니다.
이를 활용하면 좋겠군요!

TensorFlow는 더 많은 실험 결과를 공유할 수 있도록 다양한 기능을 추가하고 있습니다.

작년에 추가한 HParams Dashboard를 볼까요? 대시보드를 통해 다양한 파라미터를 직접 확인할 수 있습니다.

 

처음에 언급하였던 스크린샷 형태로 결과를 효과적으로 제공할 수 없는 문제에 대해 다시 언급하겠습니다.
TensorFlow는 이 문제를 해결하기 위해 Tensorboard dev.를 준비하였습니다.

Tensorboard dev.는 Tensorboard에서 볼 수 있는 실험 결과를 누구에게나, 쉽게 공유할 수 있도록 도와줍니다.
이를 활용하면 large-scale의 실험 결과 또한, 효율적으로 공유할 수 있습니다.

 

최근 TensorFlow를 활용한 연구 논문에서도 이를 활용하고 있습니다. 
기존에 Figure의 그림으로만 제공되던 결과를 TensorFlow dev. 링크 공유를 통해 직접 실험 결과를 확인할 수 있도록 합니다.

 

Tensorboard를 활용하는 방법은 매우 간단합니다.
케라스 콜백을 활용하세요!

model = create_model()
model.compile(optimizer='adam',
                loss='sparse_categorical_crossentropy',
                metrics=['accuracy']
                
log_dir="logs/fit/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=log_dir, histogream_freq=1)

model.fit(x=x_train, y=y_train,
          epochs=5,
          validation_data=(x_test, y_test),
          callbacks=[tensorboard_callback])

 

그리고 다음과 같이 실행하면,

!tensorboard dev upload --logdir ./logs \
  --name "My latest experiments" \
  --description "Simple comparison of several hyperparameters"

 

Tensorboard 화면을 볼 수 있습니다.

오른쪽 상단의 공유버튼을 누르고, 자신의 실험 결과를 공유하세요!