LSTM으로 시계열 데이터 예측하기

목표

장비의 cpu와 memory 지표를 수집하고 있다.
이 때 지금까지 쌓아둔 데이터를 통해 딥러닝 모델을 학습시키고, 그 모델을 통해 미래에 수집될 지표를 예측해본다.

입력 파일 준비 및 읽기

입력 파일을 형식에 맞게 읽는다. 준비된 입력 파일은 csv 파일로서 아래와 같이 되어 있다.

time,cpu.usage,memory.app
2022-01-05 00:00:00,16.3,35604
2022-01-05 00:01:00,19.9,37402
...

매 분 cpu 사용량과 memory 사용량이 수집된 데이터가 csv 파일로 저장이 되어 있다.
이 데이터를 통해 모델을 학습시켜서 최근 30분의 cpu, memory 값을 통해 다음에 수집될 cpu 사용량을 예측하도록 한다.

features = ["cpu.usage", 'memory.app']  # cpu와 memory의 값을 사용한다.
target_feature = 0  # features list에서 몇 번째의 값을 예측할지 나타낸다. cpu는 index 0에 위치한다.
num_features = len(features)
_start_time = timeit.default_timer()  # 전체 수행 시간을 기록하기 위해 timer를 설정을 한다.
dataset = read_csv(f"sample-input.csv", index_col=0)  # 파일을 읽어서 DataFrame을 반환한다. time을 index로 설정한다.
dataset.columns = features  # DataFrame의 column을 정의해준다.
raw_values = dataset.values

입력 데이터 스케일링

각 지표 값을 0에서 1 사이로 resize해서 normalize 함으로써 모델링 정확도를 높인다.

scaled value = (original value - minimum value) / (maximum value - minimum value)

지금의 예시에서 cpu는 0 ~ 100 범위의 값을 갖게 되고 memory는 수 만의 값을 갖는다. 이렇게 어떤 특성은 너무 작고 어떤 특성은 너무 크면 학습하다가 0으로 수렴하거나 발산할 수 있다.
따라서 모든 지표를 0 ~ 1로 스케일링을 함으로써 학습할 때 어느 한 특성이 과도하게 반영되는 것을 막을 수 있다.

스케일링을 할 때는 train set 데이터를 통해 scaler를 만든 후 test set 데이터는 그 scaler로 스케일링하는 것이 좋다. test set은 기대하지 못한 값이어야 하는데 학습 전부터 그 값을 scaler에 넣어버리면 처음 보는 데이터에 대한 성능을 알기 어려울 수 있기 때문이다.

# 총 샘플의 수를 구한다.
input_size = raw_values.shape[0]  

# train set을 전체의 50%, validation set을 전체의 10%로 둔다. 나머지는 test set으로 설정한다.
train_size, validation_size = int(input_size * 0.5), int(input_size * 0.6)  

# 값을 0에서 1 사이로 resize해주는 스케일러를 정의한다.
scaler = MinMaxScaler(feature_range=(0, 1))  

# train set을 기준으로 스케일러를 설정한다.
scaler.fit(raw_values[:train_size]) 

# 전체 데이터를 스케일링한다.
scaled_values = scaler.transform(raw_values)  

multi-timestep에 맞게 입력 데이터 reframe하기

지금 학습하려는 모델은 최근 10분의 데이터로 현재의 값을 에측하는 모델이다.
이전 30분의 cpu, memory 데이터를 통해서 t 시점의 cpu 값을 예측해야한다. 이를 위해서는 train set의 각 row에 입력으로 쓰일 t-30 ~ t-1의 cpu, memory 값과 출력으로 쓰일 t 시각의 cpu 값이 있어야한다.
즉, cpu(t-30), memory(t-30), cpu(t-29), memory(t-29), …, cpu(t-1), memory(t-1), cpu(t) 로 되어야한다.

# 한 번의 작업에 몇 분 기간의 input으로 예측을 할지를 정한다. t-30 ~ t-1 의 지표로 t의 지표를 예측한다.
timesteps = 30  

# reframe을 편리하게 하기 위해 우선 dataframe 타입으로 변형한다.
df = DataFrame(scaled_values)  

# 하나의 입력으로 들어갈 데이터
cols = []  
# cols의 각 column의 column name
names = []  

# 입력: t-30 ~ t-1 의 데이터가 입력으로 주어져야한다.
for i in range(timesteps, 0, -1):
        # 최근 30분의 데이터를 cols에 추가한다.
	cols.append(df.shift(i))  
        # 출력: t 시점을 예측해야한다.
	names += [('%s(t-%d)' % (f, i)) for f in features]

# 현재 데이터를 cols에 추가한다.
cols.append(df.shift(0))  
# 각 column을 옆으로 붙여준다.
names += [('%s(t)' % f) for f in features]reframed = concat(cols, axis=1)  
reframed.columns = names

# 지금의 reframed 에는 memory(t) 지표도 포함되어 있는데 cpu만 에측하려는 것이므로 이 데이터는 없애준다.
reframed.dropna(inplace=True)
num_cols = reframed.shape[1]
drop_list = []
for i in range(0, num_features):
	if i == target_feature:
		continue
        # 필요 없는 column을 계산한다.
	drop_list.append(num_cols - num_features + i)  

# 필요 없는 column을 삭제한다.
reframed.drop(reframed.columns[drop_list], axis=1, inplace=True)  

train set, validation set, test set 나누기

reframed 된 데이터를 통해서 train set, validation set, test set을 나눈다.

• train set: 모델을 학습하는 데에 쓰이는 데이터이다.
• validation set: 모델을 학습할 때 검증되는 데에 쓰이는 데이터이다. train set을 통해 모델을 학습시키면 그 모델을 validation set으로 테스트해봐서 어느 정도의 정확도가 나오는지 확인한다. train set으로만 학습하면 overfitting의 우려가 있는데 이를 막아줄 수 있다.
• test set: 학습이 완료된 모델로 실제 테스트를 할 데이터이다.
  reframe하면서 전체 row 수가 줄어들 수 있다. 따라서 reframed[:train_size, :]가 처음에 생각했던 정확한 50%가 아닐 수 있지만 작은 오차이고 전체 동작에 유의미한 영향을 주지 않으므로 무시한다.

values = reframed.values
train, validation, test = values[:train_size, :], values[train_size:validation_size, :], values[validation_size:, :]

# 입력과 출력을 나눈다. x를 입력, y를 출력이라고 할 때 각 row에서 마지막 하나 빼고는 x이고 마지막 하나가 y이다.
train_x, train_y = train[:, :-1], train[:, -1:]
validation_x, validation_y = validation[:, :-1], validation[:, -1:]
test_x, test_y = test[:, :-1], test[:, -1:]

RNN 에 맞게 데이터 reframe하기

일반적인 MLP에서는 (size, feature)의 2차원이지만 RNN은 시간 개념이 있기 때문에 3차원의 인풋을 받는다.
(batch_size, timesteps, input_dim)

train_x = train_x.reshape((train_x.shape[0], timesteps, num_features))
validation_x = validation_x.reshape((validation_x.shape[0], timesteps, num_features))
test_x = test_x.reshape((test_x.shape[0], timesteps, num_features))

LSTM 모델 정의하고 train set, validation set으로 학습 진행하기

Sequential 를 통해 레이어를 선형으로 연결하여 구성하는 모델을 정의한다. add로 레이어를 추가할 수 있다.
input layer 수는 여러 값을 테스트해보면서 최적을 값을 찾아가야하는데 우선은 10개로 해본다.
input shape는 입력의 모습을 알려주는 정보인데 첫 번째 레이어에만 알려주면 된다.
마지막에는 Dense layer를 추가해주는데 뉴런의 출력을 연결해주는 역할이다. Dense layer는 입력 뉴런 수에 관계 없이 출력 뉴런 수를 자연스럽게 설정할 수 있어 출력층으로 많이 사용된다. 예측하고자하는 target은 현재 시점의 cpu 값 한 개이므로 output은 1로 설정한다.

model = Sequential()
model.add(LSTM(10, input_shape=(train_x.shape[1], train_x.shape[2])))
model.add(Dense(1))

모델을 학습시키기 이전에, compile 메소드를 통해서 학습 방식에 대한 환경설정을 해야한다.
Loss function은 모델이 최적화할 때 사용되는 함수로서 이 값을 최소화시키는 방향으로 학습된다. (참고: https://neptune.ai/blog/keras-loss-functions)

• mse: large error를 penalize하고 싶을 때 쓴다. 아웃라이어에 약하다.
• mape: loss가 직관적일 때 쓰고 아웃라이어에 강하다.

Optimizer는 loss function을 통해 얻은 손실값으로부터 모델을 업데이트하는 방식을 의미한다.
Loss function, optimizer도 여려 개를 사용해보면서 모델에 적합한 것을 찾는 것이 좋다. 우선은 mae와 adma으로 설정해본다.

model.compile(loss='mae', optimizer='adam')
model.summary()

모델을 학습할 때 loss가 감소하지 않으면 fitting을 멈춘다.

• fit network: compile에서 지정한 방식대로 학습을 수행하여 weight를 찾고 metric을 반환한다.
• epoch: 데이터를 한 번 학습하는 것을 몇 번 반복할지를 정한다. 적당한 epoch 값을 골라야 overfitting과 underfitting을 막을 수 있다.
• batch size: 메모리의 한계와 속도 저하 때문에 대부분의 경우에는 한 번의 epoch에서 모든 데이터를 한꺼번에 집어넣을 수는 없다. 따라서 데이터를 나누어서 주게 되는데 이때 몇 번 나누어서 주는가를 iteration, 각 iteration마다 주는 데이터 사이즈를 batch size라고 한다.

train 데이터로 학습을 하게 되는데 매 epoch 이 끝나게 되면 validation 데이터로 모델 정확도를 검증한다.
epoch을 반복함에 따라 모델 정확도가 높아져야하는데 어느 순간부터는 train 데이터에 너무 종속되어 overfitting이 발생할 수 있다.
그렇게 되면 validation 데이터에 대해서는 정확도가 떨어지기 때문에 validation 데이터에 대한 정확도가 떨어질 때 학습을 멈추는 것이 좋다.
이를 위해 EarlyStopping을 정의해준다. validation set에 대한 loss를 의미하는 val_loss 를 모니터 대상으로 두고 10회 이상 정확도가 감소한다면 설정된 epoch 수만큼 학습을 안 하더라도 멈추도록 하는 것이다.

fit의 결과로 나오는 history에는 각 epoch의 loss, val_loss 가 저장되어 있다.

early_stop = EarlyStopping(monitor='val_loss', patience=10, verbose=1)
history = model.fit(train_x, train_y, epochs=1, validation_data=(validation_x, validation_y), batch_size=100, verbose=1, callbacks=[early_stop], shuffle=False)
history_loss = history.history['loss']pyplot.figure()
pyplot.plot(history.history['loss'], label='train_loss')
pyplot.plot(history.history['val_loss'], label='validation_loss')
pyplot.legend()
pyplot.draw()

학습된 모델로 test set을 입력하여 예측된 결과 얻기

predict 함수를 통해서 test_x의 각 입력에 대한 다음 시각의 예측된 cpu 값을 numpy array로 받을 수 있다.
이 결과값은 0에서 1 사이로 scaled 된 상태의 결과이므로 inverse하여 원래의 값을 얻도록 한다.
처음에 scaling할 때 (cpu, memory)의 형태로 했었기 때문에 그 형태로 맞춰야한다.
지금의 결과값은 (cpu) column만 있는데 dummy column을 뒤에 하나 붙여줘서 inverse_transform을 한다.
그 결과의 첫 번째 column만 받아오면 원하는 결과값이 된다.

test_yhat = model.predict(test_x)

test_yhat = concatenate((test_yhat, test_yhat), axis=1)
test_yhat = scaler.inverse_transform(test_yhat)
test_yhat = test_yhat[:, :1]

test_y = concatenate((test_y, test_y), axis=1)
test_y = scaler.inverse_transform(test_y)
test_y = test_y[:, :1]

print(f"test_yhat: \n{test_yhat}")
print(f"test_y: \n{test_y}")

pyplot.figure()
pyplot.plot(list(map(lambda x: x[0], test_y)), label=f'Actual {features[target_feature]} usage')
pyplot.plot(list(map(lambda x: x[0], test_yhat)), label=f'Predicted {features[target_feature]} usage')
pyplot.tight_layout()
pyplot.legend()
pyplot.draw()

_end_time = timeit.default_timer()
print(f"\nElapsed Time: {_end_time - _start_time}\n")
pyplot.show()