---
html:
toc: true
offline: true
export_on_save:
html: true
---
# KSB 프레임워크와 연동을 위한 tensorflow 학습코드 작성하기
---
KSB 인공지능 프레임워크는 파이썬 기반 사용자의 tensorflow 코드로 파라미터를 전달하고, 텐서플로우 모델 서빙용으로 export된 모델을 하둡파일시스템(HDFS)으로 복사해오는 방식으로 연동이 됩니다.
따라서 사용자는 tensorflow 모델 학습을 위한 코드 작성 시, 프레임워크에서 전달하는 파라미터(입력파일 경로, 모델 저장 경로 등)를 받는 부분, 학습된 모델을 텐서플로우 서빙용으로 export 하는 부분을 구현해야 합니다.
KSB 인공지능 프레임워크에서 텐서플로우 기반 모델을 학습하고 서빙하는 절차는 아래와 같습니다. **교통속도 예측 튜토리얼** 을 참고합니다.
1. KSB 인공지능 프레임워크에서 모델 입력 데이터 (모델 출력 데이터, 즉 lable 포함) 를 가공하여 csv 파일로 저장합니다. 하나의 파일에 입력 데이터와 학습을 위한 target 이 같이 저장되어야 합니다. 교통속도 스트림 데이터 전처리하기 를 참고합니다.
2. 사용자 파이썬 코드에서 앞에서 가공한 csv 파일을 읽고 딥러닝 모델 학습을 합니다. 프레임워크와 사용자 파이썬 코드는 입력 파일 경로, 모델 export 경로 등의 파라미터를 전달할 수 있도록 연동되어 있습니다. 교통속도예측 텐서플로우 모델 학습하기 를 참고합니다.
3. 학습 후 export 한 모델을 프레임워크에서 REST API 서비스가 가능하도록 띄우고, 클라이언트는 쿼리를 통해 예측 결과를 받을 수 있습니다. 사용자는 온디맨드 방식으로 서비스를 구성하거나 (온디맨드 방식으로 교통속도예측 모델 서빙하기 참고) 스트림 방식으로 서비스를 구성할 수 있습니다 (스트림 방식으로 교통속도예측 모델 서빙하기 참고).
## 프레임워크에서 전달하는 파라미터를 받는 부분 작성
사용자 tensorflow 코드를 KSB 인공지능 프레임워크와 연동하기 위해서는 기본적으로 4개의 path를 KSB 인공지능 프레임워크로부터 전달 받아야 합니다. 그 밖에 사용자가 필요한 파라미터도 선언하여 전달 받을 수 있습니다.
```python
import argparse
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Train a model on tensorflow.')
## 기본적인 4개의 argument 선언
parser.add_argument('--input', type=str, default='traffic_processing.csv', help='input path')
parser.add_argument('--model', type=str, default='file:///home/csle/kangnam/model/0000',
help='model path')
parser.add_argument('--modelPath', type=str, default='file:///home/csle/kangnam/model',
help='model base path')
parser.add_argument('--output', type=str, default='file:///home/csle/kangnam/output',
help='output path')
## 사용자가 필요한 argument 선언
parser.add_argument('--isTrain', type=bool, default=True,
help='If true, train model. If not, predict with model')
parser.add_argument('--num_epochs', type=int, default=3, help='# of epochs')
## FLAGS 는 전역 변수
## FLAGS 변수를 사용해서 파라미터 사용
FLAGS, unparsed = parser.parse_known_args()
```
KSB 인공지능 프레임워크에서 전달하는 파라미터는 웹툴킷에서 설정한 컴퍼넌트의 속성 값입니다.
### --input : FileReader의 filePath
FileReader의 속성 중 filePath 에 입력 파일의 경로 및 파일명을 지정합니다. 이 값은 사용자 tensorflow 코드의 --input 파라미터로 전달됩니다.
아래와 같이 3가지 경로가 입력 가능합니다.
1) 로컬 파일시스템 절대경로
file:///home/csle/ksb-csle/examples/input/{선택한 경로 및 파일명}
2) hdfs 절대경로
hdfs://{host}:{port}/user/{user_id}/dataset/{선택한 경로 및 파일명}
3) Hdfs 상대경로
dataset/{선택한 경로 및 파일명}
### --output : FileWriter의 filePath
FileWriter의 속성 중 filePath에 모델이 저장될 경로를 지정합니다. 이 값은 사용자 tensorflow 코드의 --output 파라미터로 전달됩니다. 폴더명까지만 입력합니다. 해당 폴더에 자동으로 최신버전의 폴더를 생성한 후 모델이 저장됩니다.
아래와 같이 3가지 경로가 입력 가능합니다.
1) 로컬 파일시스템 절대경로
file:///home/csle/ksb-csle/{선택한 경로}
2) hdfs 절대경로
hdfs://{host}:{port}/user/{user_id}/model/{선택한 경로}
3) hdfs 상대경로
model/{선택한 경로}
위와 같이 입력하는 경우, 자동으로 **사용자 기본 파일경로** 가 앞에 추가됩니다. 따라서 아래와 같은 위치에 모델이 저장됩니다. hdfs://{host}:{port}/user/{user_id}/applications/{workflow_id}/output/model/{선택한 경로}
### --modelPath (--model 포함) : DLTrainOperator의 modelPath
DLTrainOperator의 속성 중 modelPath에 내부적으로 모델을 저장할 경로를 지정합니다. 이 값은 사용자 tensorflow 코드의 --modelPath 파라미터로 전달됩니다. 폴더명까지만 입력합니다.
아래와 같이 로컬 file system만 지원합니다.
1) 로컬 파일시스템 절대경로
file:///home/csle/ksb-csle/{선택한 경로}
--model 파라미터는 위에서 지정한 modelPath를 기준으로 모델버전 폴더 path를 자동생성하여 전달합니다. 모델 버전은 0000 부터 1씩 자동으로 증가합니다.
- (modelPath) file:///home/csle/ksb-csle/{선택한 경로}
- (자동생성되는 model) file:///home/csle/ksb-csle/{선택한 경로}/{모델버전}
### 그 밖의 파라미터 : DLTrainOperator의 additionalParams
DLTrainOperator의 속성 중 additionalParams에 사용자가 필요한 파라미터의 이름 및 값을 설정합니다.
paramName 에 파라미터 이름을 입력하고, paramValue 에 전달할 값을 입력합니다.
## 입력파일을 전처리 하는 부분 작성
KSB 인공지능 프레임워크에서 전달한 입력파일로부터 텐서플로우 모델의 배치 입력 데이터를 생성하기 쉽도록 path를 자동으로 획득하기 위한 pathResolver package 및 DataLoader 클래스를 제공합니다.
```python
## import pathresolver package
import pathResolver as pathRes
def main():
# Preparation of data.
pathresolver = pathRes.PathResolver(FLAGS.input, FLAGS.modelPath,
FLAGS.model, FLAGS.output)
_, local_model_path, local_checkpoint_file, _ = pathresolver.get_paths()
data = pathresolver.get_input_dataframe()
loader = DataLoader(data)
```
KSB 인공지능 프레임워크로부터 받은 4가지 path 정보를 이용하여 PathResolver 인스턴스를 생성합니다.
```python
## Initiate pathresolver
pathresolver = pathRes.PathResolver(FLAGS.input, FLAGS.modelPath, FLAGS.model, FLAGS.output)
```
get_paths 함수를 호출하여 4개의 path를 리턴받습니다.
local_model_path 위치에 tensorflow serving용 모델을 저장합니다. 그리고 local_checkpoint_file 위치에 텐서플로우 모델의 체크포인트 파일을 저장합니다. local_model_base_path 및 output_file_path 는 참고용으로 사용하지 않습니다.
```python
## Get paths from pathresolver
local_model_base_path, local_model_path, local_checkpoint_file, output_file_path = \
pathresolver.get_paths()
```
입력 파일로부터 pandas dataframe을 리턴받습니다. 그리고 DataLoader 클래스를 이용하여 입력 데이터를 가공합니다.
```python
## get dataframe from input path
data = pathresolver.get_input_dataframe()
loader = DataLoader(data)
```
```python
class DataLoader:
def __init__(self, data):
def prepare_data(self):
"""
Prepare data for training a graph.
Returns:
Input/output data in numpy array.
"""
self.inputs = input #모델 입력 데이터 변수
self.outputs = output #모델 출력 데이터 변수
def get_batch(self, batch_size=64, shuffle=True):
```
prepare_data 함수에서 사용자의 텐서플로우 모델 입력에 적합한 데이터로 가공하는 코드를 작성합니다. 모델의 입력 데이터는 self.inputs 변수로 저장하고, 모델의 출력 데이터는 self.outputs 변수로 저장합니다.
## 텐서플로우 모델을 학습하고, 학습된 모델을 서빙용으로 export 하는 부분 작성
텐서플로우 모델을 정의하고 학습한 후, 학습된 모델을 텐서플로우 서빙용으로 export 하는 코드를 작성합니다. TensorFlowModel 클래스의 각 함수 부분에 필요한 코드를 작성합니다.
```python
tf_model = TensorFlowModel(local_model_path, local_checkpoint_file)
tf_model.build_graph()
tf_model.train_graph(loader)
# Store model to target Hdfs
pathresolver.store_output_model()
```
```python
class TensorFlowModel():
def __init__(self, local_model_path, local_checkpoint_file):
def build_graph(self):
def train_graph(self, loader):
# checkpoint save
saver.save(sess, self.local_checkpoint_file)
# tensorflow serving model export
self.export(sess)
def export(self, session):
```
위에서 할당받은 local_checkpoint_file 위치에 모델 checkpoint 파일을 저장합니다.
```python
saver.save(sess, self.local_checkpoint_file)
```
위에서 할당받은 local_model_path 위치에 tensorflow serving용 모델을 저장합니다. 텐서플로우 그래프의 입력 tensor 이름, 출력 tensor 이름, signature 이름 등을 정의합니다.
```python
# tensorflow serving model export
self.export(sess)
```
```python
def export(self, session):
"""
Export a model.
"""
export_dir_path = self.local_model_path
if os.path.exists(export_dir_path):
shutil.rmtree(export_dir_path)
## 모델 export path 받기
builder = tf.saved_model.builder.SavedModelBuilder(export_dir_path)
signature = tf.saved_model.signature_def_utils.build_signature_def(
## 그래프 input tensor 정의
inputs={
"in1": tf.saved_model.utils.build_tensor_info(self.X)
},
## 그래프 output tensor 정의
outputs={
"out1": tf.saved_model.utils.build_tensor_info(self.output)
},
method_name=tf.saved_model.signature_constants.PREDICT_METHOD_NAME
)
builder.add_meta_graph_and_variables(
session,
tags=[tf.saved_model.tag_constants.SERVING],
signature_def_map={"predict_speed": signature}, ## Signature 이름 정의
assets_collection=None,
legacy_init_op=None,
clear_devices=None,
main_op=None)
builder.save()
print('exported model in ', export_dir_path)
```
output_file_path에 모델을 저장합니다.
```python
## Store model to target file system
pathresolver.store_output_model()
```
## 전체 코드 예제
15분 뒤 교통속도를 예측하는 텐서플로우 모델을 구현한 예제 입니다. 기본적인 LSTM 을 사용하여 구현하였습니다. 프레임워크에서 코드를 연동하여 실행하는 방법은 교통속도예측 텐서플로우 모델 학습하기 를 참고합니다.
main 함수를 포함한 사용자 코드에서 사용하는 라이브러리 파일 등은 동일한 폴더에 위치 시키거나 1 depth의 하위 폴더까지 작성 가능합니다.
```python
import os
import sys
import shutil
import argparse
import numpy as np
import pandas as pd
import tensorflow as tf
import pathResolver as pathRes
class DataLoader:
def __init__(self, data):
"""
DataLoader class.
Args:
data (pandas.DataFrame): Data in Pandas DataFrame format.
"""
self.data = data
# User-customized data.
self.inputs = None
self.outputs = None
self.prepare_data()
def prepare_data(self):
"""
Prepare data for training a graph.
Returns:
Input/output data in numpy array.
"""
# Convert the first column to datetime type.
self.data.ix[:, 0] = pd.to_datetime(self.data.ix[:, 0])
# Sort by the date.
self.data = self.data.sort_values(0)
# Get 8 rows as input, and set the row three after as output.
data_np = self.data.values[::, 1:]
inputs = []
outputs = []
for i in range(len(data_np) - 8 - 3):
input = data_np[i:i + 8].flatten()
output = data_np[i + 8 + 3].flatten()
inputs.append(input)
outputs.append(output)
inputs = np.array(inputs)
outputs = np.array(outputs)
inputs = inputs.reshape(-1, 8, 170)
self.inputs = inputs
self.outputs = outputs
def get_batch(self, batch_size=64, shuffle=True):
"""
Get batch samples of the shuffled dataset.
Args:
batch_size (int): Batch size.
shuffle (boolean): If shuffle or not.
Returns:
Sample of batch size.
"""
# Shuffle data.
if shuffle:
shuffle_index = np.arange(len(self.inputs))
np.random.shuffle(shuffle_index)
shuffled_inputs = self.inputs[shuffle_index]
shuffled_outputs = self.outputs[shuffle_index]
else:
shuffled_inputs = self.inputs
shuffled_outputs = self.outputs
# Enumerate.
n_iter = int(shuffled_inputs.shape[0] / batch_size)
for i in range(n_iter):
yield shuffled_inputs[i * batch_size: (i + 1) * batch_size], \
shuffled_outputs[i * batch_size: (i + 1) * batch_size]
class TensorFlowModel():
def __init__(self, local_model_path, local_checkpoint_file):
"""
Build graph and export a model.
"""
self.X = None
self.Y = None
self.loss = None
self.train = None
self.output = None
self.local_model_path = local_model_path
self.local_checkpoint_file = local_checkpoint_file
def build_graph(self):
"""
Build a graph
"""
# Define inputs and outputs
seq_length = 8
num_links = 170
hidden_size = 128
learning_rate = 0.01
self.X = tf.placeholder(tf.float32, shape = [None, seq_length, num_links])
self.Y = tf.placeholder(tf.float32, shape = [None, num_links])
# build a LSTM network
cell = tf.contrib.rnn.BasicLSTMCell(
num_units=hidden_size, state_is_tuple=True, activation=tf.tanh)
outputs, _states = tf.nn.dynamic_rnn(cell, self.X, dtype=tf.float32)
Y_pred = tf.contrib.layers.fully_connected(
outputs[:, -1], num_links, activation_fn=None) # We use the last cell's output
# cost/loss
self.loss = tf.reduce_sum(tf.square(Y_pred - self.Y)) # sum of the squares
# optimizer
optimizer = tf.train.AdamOptimizer(learning_rate)
self.train = optimizer.minimize(self.loss)
self.output = Y_pred
def train_graph(self, loader):
"""
Train a graph
"""
saver = tf.train.Saver()
with tf.Session() as sess:
init = tf.global_variables_initializer()
sess.run(init)
# restore from checkpoint file
if os.path.exists(self.local_checkpoint_file):
ckpt_path = tf.train.latest_checkpoint(self.local_checkpoint_file)
saver.restore(sess, ckpt_path)
print("Model Restored from ", ckpt_path)
num_epochs = FLAGS.num_epochs
num_batch = 10
for i in range(num_epochs):
batch_loader = loader.get_batch(batch_size=num_batch, shuffle=False)
for inputs, outputs in batch_loader:
_, loss = sess.run([self.train, self.loss],
feed_dict={self.X: inputs, self.Y: outputs})
print("train lose: ", loss)
# checkpoint save
saver.save(sess, self.local_checkpoint_file)
# tensorflow serving model export
self.export(sess)
sess.close()
tf.reset_default_graph()
def export(self, session):
"""
Export a model.
"""
export_dir_path = self.local_model_path
if os.path.exists(export_dir_path):
shutil.rmtree(export_dir_path)
builder = tf.saved_model.builder.SavedModelBuilder(export_dir_path)
signature = tf.saved_model.signature_def_utils.build_signature_def(
inputs={
"in1": tf.saved_model.utils.build_tensor_info(self.X)
},
outputs={
"out1": tf.saved_model.utils.build_tensor_info(self.output)
},
method_name=tf.saved_model.signature_constants.PREDICT_METHOD_NAME
)
builder.add_meta_graph_and_variables(
session,
tags=[tf.saved_model.tag_constants.SERVING],
signature_def_map={"predict_speed": signature},
assets_collection=None,
legacy_init_op=None,
clear_devices=None,
main_op=None)
builder.save()
print('exported model in ', export_dir_path)
def main():
# Preparation of data.
pathresolver = pathRes.PathResolver(FLAGS.input, FLAGS.modelPath,
FLAGS.model, FLAGS.output)
_, local_model_path, local_checkpoint_file, _ = pathresolver.get_paths()
data = pathresolver.get_input_dataframe()
loader = DataLoader(data)
if FLAGS.isTrain:
# Train a model and export.
tf_model = TensorFlowModel(local_model_path, local_checkpoint_file)
tf_model.build_graph()
tf_model.train_graph(loader)
# Store model to target Hdfs
pathresolver.store_output_model()
else:
pass
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Train a model on tensorflow.')
parser.add_argument('--input', type=str, default='traffic_processing.csv', help='input path')
parser.add_argument('--model', type=str, default='file:///home/csle/kangnam/model/0000',
help='model path')
parser.add_argument('--modelPath', type=str, default='file:///home/csle/kangnam/model',
help='model base path')
parser.add_argument('--output', type=str, default='file:///home/csle/kangnam/output',
help='output path')
parser.add_argument('--isTrain', type=bool, default=True,
help='If true, train model. If not, predict with model')
parser.add_argument('--num_epochs', type=int, default=3, help='# of epochs')
FLAGS, unparsed = parser.parse_known_args()
main()
```