블로그 이미지
평범하게 살고 싶은 월급쟁이 기술적인 토론 환영합니다.같이 이야기 하고 싶으시면 부담 말고 연락주세요:이메일-bwcho75골뱅이지메일 닷컴. 조대협


Archive»


 
 

텐서플로우의 세션,그래프 그리고 함수의 개념


조대협 (http://bcho.tistory.com)


그래프와 세션에 대한 개념이 헷갈려서, 좋은 샘플이 하나 만들어져서 공유합니다.

텐서 플로우의 기본 작동 원리는 세션 시작전에 그래프를 정의해놓고, 세션을 시작하면 그 그래프가 실행되는 원리인데, 그래서 이 개념이 일반적인 프로그래밍 개념과 상의하여 헷갈리는 경우가 많다


즉, 세션을 시작해놓고 함수를 호출하는 케이스들이 대표적인데

http://bcho.tistory.com/1170 코드를 재 사용해서 이해해보도록 하자


이 코드를 보면, tt = time * 10 을 세션 시작전에 정의해놨는데, 이 코드를 함수로 바꾸면 아래와 같은 형태가 된다.


변경전 코드

def main():

   

   print 'start session'

   #coornator 위에 코드가 있어야 한다

   #데이타를 집어 넣기 전에 미리 그래프가 만들어져 있어야 함.

   batch_year,batch_flight,batch_time = read_data_batch(TRAINING_FILE)

   year = tf.placeholder(tf.int32,[None,],name='year')

   flight = tf.placeholder(tf.string,[None,],name='flight')

   time = tf.placeholder(tf.int32,[None,],name='time')

   

   tt = time * 10

   summary = tf.summary.merge_all()

   with tf.Session() as sess:

       summary_writer = tf.summary.FileWriter(LOG_DIR,sess.graph)

       try:


           coord = tf.train.Coordinator()

           threads = tf.train.start_queue_runners(sess=sess, coord=coord)


           for i in range(5):

               y_,f_,t_ = sess.run([batch_year,batch_flight,batch_time])

               print sess.run(tt,feed_dict={time:t_})

               #summary_str = sess.run(summary,feed_dict=feed_dict)

               #summary_writer.add_summary(summary_str,i)

               summary_writer.flush()         


변경후 코드

def create_graph(times):

   tt = times * 10

   return tt


def main():

   

   print 'start session'

   #coornator 위에 코드가 있어야 한다

   #데이타를 집어 넣기 전에 미리 그래프가 만들어져 있어야 함.

   batch_year,batch_flight,batch_time = read_data_batch(TRAINING_FILE)

   year = tf.placeholder(tf.int32,[None,],name='year')

   flight = tf.placeholder(tf.string,[None,],name='flight')

   time = tf.placeholder(tf.int32,[None,],name='time')

   

   r = create_graph(time)

   

   summary = tf.summary.merge_all()

   with tf.Session() as sess:

       summary_writer = tf.summary.FileWriter(LOG_DIR,sess.graph)

       try:


           coord = tf.train.Coordinator()

           threads = tf.train.start_queue_runners(sess=sess, coord=coord)


           for i in range(5):

               y_,f_,t_ = sess.run([batch_year,batch_flight,batch_time])

               print sess.run(r,feed_dict={time:t_})

               #summary_str = sess.run(summary,feed_dict=feed_dict)

               #summary_writer.add_summary(summary_str,i)

               summary_writer.flush()


변경후 코드는 tt = times * 10 을 create_graph라는 함수로 뺐는데, session 시작전에 함수를 호출한다. 언뜻 보면 개념이 헷갈릴 수 있는데, time 이라는 변수는 텐서플로우의 placeholder로 값이 읽혀지는 시점이 queue_runner를 시작해야 값을 읽을 수 있는 준비 상태가 되고, 실제로 값을 큐에서 읽으려면 session을 실행하고 feed_dict를 이용하여 feeding을 해줘야 값이 채워지기 때문에, 일반적인 프로그램상으로는 session을 시작한 후에 함수를 호출해야할것 같이 생각이 되지만, 앞에서도 언급했듯이 텐서플로우에서 프로그래밍의 개념은 그래프를 다 만들어놓은 후 (데이타가 처리되는 흐름을 모두 정의해놓고) 그 다음 session을 실행하여 그래프에 데이타를 채워놓는 개념이기 때문에, session이 정의되기 전에 함수 호출등을 이용해서 그래프를 정의해야 한다.



텐서플로우 배치 처리


조대협 (http://bcho.tistory.com)


텐서플로우에서 파일에서 데이타를 읽은 후에, 배치처리로 placeholder에서 읽는  예제를 설명한다.

텐서의 shape 의 차원과 세션의 실행 시점등이 헷갈려서 시행착오가 많았기 때문에 글로 정리해놓는다.

큐와 파일처리에 대한 기본적인 내용은 아래글

  • http://bcho.tistory.com/1163

  • http://bcho.tistory.com/1165

를 참고하기 바란다.

데이타 포맷

읽어 드릴 데이타 포맷은 다음과 같다. 비행기 노선 정보에 대한 데이타로 “년도,항공사 코드, 편명"을 기록한 CSV 파일이다.

2014,VX,121

2014,WN,1873

2014,WN,2787

배치 처리 코드

이 데이타를 텐서 플로우에서 읽어서 배치로 place holder에 feeding 하는 코드 이다

먼저 read_data는 csv 파일에서 데이타를 읽어서 파싱을 한 후 각 컬럼을 year,flight,time 으로 리턴하는 함 수이다.

def read_data(file_name):

   try:

       csv_file = tf.train.string_input_producer([file_name],name='filename_queue')

       textReader = tf.TextLineReader()

       _,line = textReader.read(csv_file)

       year,flight,time = tf.decode_csv(line,record_defaults=[ [1900],[""],[0] ],field_delim=',')    

   except:

       print "Unexpected error:",sys.exc_info()[0]

       exit()

   return year,flight,time


string_input_producer를 통해서 파일명들을 큐잉해서 하나씩 읽는데,여기서는 편의상 하나의 파일만 읽도록 하였는데, 여러개의 파일을 병렬로 처리하고자 한다면, [file_name]  부분에 리스트 형으로 여러개의 파일 목록을 지정해주면 된다.

다음 각 파일을 TextReader를 이용하여 라인 단위로 읽은 후 decode_csv를 이용하여, “,”로 분리된 컬럼을 각각  읽어서 year,flight,time 에 저장하여 리턴하였다.


다음 함수는 read_data_batch 라는 함수인데, 앞에서 정의한 read_data 함수를 호출하여, 읽어드린 year,flight,time 을 배치로 묶어서 리턴하는 함수 이다.


def read_data_batch(file_name,batch_size=10):

   year,flight,time = read_data(file_name)

   batch_year,batch_flight,batch_time = tf.train.batch([year,flight,time],batch_size=batch_size)

   

   return  batch_year,batch_flight,batch_time


tf.train.batch 함수가 배치로 묶어서 리턴을 하는 함수인데, batch로 묶고자 하는 tensor 들을 인자로 준 다음에, batch_size (한번에 묶어서 리턴하고자 하는 텐서들의 개수)를 정해주면 된다.


위의 예제에서는 batch_size를 10으로 해줬기 때문에, batch_year = [ 1900,1901….,1909]  와 같은 형태로 10개의 년도를 하나의 텐서에 묶어서 리턴해준다.

즉 입력 텐서의 shape이  [x,y,z] 일 경우 tf.train.batch를 통한 출력은 [batch_size,x,y,z] 가 된다.(이 부분이 핵심)


메인 코드

자 이제 메인 코드를 보자

def main():

   

   print 'start session'

   #coornator 위에 코드가 있어야 한다

   #데이타를 집어 넣기 전에 미리 그래프가 만들어져 있어야 함.

   batch_year,batch_flight,batch_time = read_data_batch(TRAINING_FILE)

   year = tf.placeholder(tf.int32,[None,])

   flight = tf.placeholder(tf.string,[None,])

   time = tf.placeholder(tf.int32,[None,])

   

   tt = time * 10


tt = time * 10 이라는 공식을 실행하기 위해서 time 이라는 값을 읽어서 피딩하는 예제인데 먼저 read_data_batch를 이용하여 데이타를 읽는 그래프를 생성한다. 이때 주의해야할점은 이 함수를 수행한다고 해서, 바로 데이타를 읽기 시작하는 것이 아니라, 데이타의 흐름을 정의하는 그래프만 생성된다는 것을 주의하자


다음으로는 year,flight,time placeholder를 정의한다.

year,flight,time 은 0 차원의 scalar 텐서이지만, 값이 연속적으로 들어오기 때문에, [None, ] 로 정의한다.

즉  year = [1900,1901,1902,1903,.....] 형태이기 때문에 1차원 Vector 형태의 shape으로 [None, ] 로 정의한다.

Placeholder 들에 대한 정의가 끝났으면, 세션을 정의하고 데이타를 읽어드리기 위한 Queue runner를 수행한다. 앞의 과정까지 텐서 그래프를 다 그렸고, 이 그래프 값을 부어넣기 위해서, Queue runner 를 수행한 것이다.


   with tf.Session() as sess:

       try:


           coord = tf.train.Coordinator()

           threads = tf.train.start_queue_runners(sess=sess, coord=coord)


Queue runner를 실행하였기 때문에 데이타가 데이타 큐로 들어가기 시작하고, 이 큐에 들어간 데이타를 읽어드리기 위해서, 세션을 실행한다.

               y_,f_,t_ = sess.run([batch_year,batch_flight,batch_time])

               print sess.run(tt,feed_dict={time:t_})

세션을 실행하면, batch_year,batch_flight,batch_time 값을 읽어서 y_,f_,t_ 변수에 각각 집어 넣은 다음에, t_ 값을 tt 공식의 time 변수에 feeding 하여, 값을 계산한다.


모든 작업이 끝났으면 아래와 같이 Queue runner를 정지 시킨다.

           coord.request_stop()

           coord.join(threads)


다음은 앞에서 설명한 전체 코드이다.


import tensorflow as tf

import numpy as np

import sys


TRAINING_FILE = '/Users/terrycho/dev/data/flight.csv'


## read training data and label

def read_data(file_name):

   try:

       csv_file = tf.train.string_input_producer([file_name],name='filename_queue')

       textReader = tf.TextLineReader()

       _,line = textReader.read(csv_file)

       year,flight,time = tf.decode_csv(line,record_defaults=[ [1900],[""],[0] ],field_delim=',')    

   except:

       print "Unexpected error:",sys.exc_info()[0]

       exit()

   return year,flight,time


def read_data_batch(file_name,batch_size=10):

   year,flight,time = read_data(file_name)

   batch_year,batch_flight,batch_time = tf.train.batch([year,flight,time],batch_size=batch_size)

   

   return  batch_year,batch_flight,batch_time


def main():

   

   print 'start session'

   #coornator 위에 코드가 있어야 한다

   #데이타를 집어 넣기 전에 미리 그래프가 만들어져 있어야 함.

   batch_year,batch_flight,batch_time = read_data_batch(TRAINING_FILE)

   year = tf.placeholder(tf.int32,[None,])

   flight = tf.placeholder(tf.string,[None,])

   time = tf.placeholder(tf.int32,[None,])

   

   tt = time * 10


   with tf.Session() as sess:

       try:


           coord = tf.train.Coordinator()

           threads = tf.train.start_queue_runners(sess=sess, coord=coord)


           for i in range(10):

               y_,f_,t_ = sess.run([batch_year,batch_flight,batch_time])

               print sess.run(tt,feed_dict={time:t_})


           print 'stop batch'

           coord.request_stop()

           coord.join(threads)

       except:

           print "Unexpected error:", sys.exc_info()[0]


main()


다음은 실행결과이다.



텐서플로우-#1 자료형의 이해

빅데이타/머신러닝 | 2016.12.09 22:42 | Posted by 조대협

텐서플로우-#1 자료형의 이해


조대협 (http://bcho.tistory.com)


딥러닝에 대한 대략적인 개념을 익히고 실제로 코딩을 해보려고 하니, 모 하나를 할때 마다 탁탁 막힌다. 파이썬이니 괜찮겠지 했는데, (사실 파이썬도 다 까먹어서 헷갈린다.) 이건 라이브러리로 도배가 되어 있다.

당연히 텐서플로우 프레임웍은 이해를 해야 하고, 데이타를 정재하고 시각화 하는데, numpy,pandas와 같은 추가적인 프레임웍에 대한 이해가 필요하다.


node.js 시작했을때도 자바스크립트 때문에 많이 헤매고 몇달이 지난후에야 어느정도 이해하게 되었는데, 역시나 차근차근 기초 부터 살펴봐야 하지 않나 싶다.


텐서 플로우에 대해 공부한 내용들을 하나씩 정리할 예정인데, 이 컨텐츠들은 유투브의 이찬우님의 강의를 기반으로 정리하였다. 무엇보다 한글이고 개념을 쉽게 풀어서 정리해주시기 때문에, 왠만한 교재 보다 났다.

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


텐서플로우 환경 설정

텐서 플로우 환경을 설정 하는 방법은 쉽지 않다. 텐서플로우 뿐 아니라, 여러 파이썬 버전과 그에 맞는 라이브러리도 함께 설정해야 하기 때문에 여간 까다로운게 아닌데, 텐서플로우 환경은 크게 대략 두 가지 환경으로 쉽게 설정이 가능하다.

구글 데이타랩

첫번째 방법은 구글에서 주피터 노트북을 도커로 패키징해놓은 패키지를 이용하는 방법이다. 도커 패키지안에, numpy,pandas,matplotlib,tensorflow,python 등 텐서플로우 개발에 필요한 모든 환경이 패키징 되어 있다. 데이타 랩 설치 방법은 http://bcho.tistory.com/1134 링크를 참고하면 된다.

도커 런타임이 설치되어 있다면, 데이타랩 환경 설정은 10분이면 충분하다.

아나콘다

다음 방법은 일반적으로 가장 많이 사용하는 방법인데, 파이썬 수학관련 라이브러리를 패키징해놓은 아나콘다를 이용하는 방법이 있다. 자세한 환경 설정 방법은 https://www.tensorflow.org/versions/r0.12/get_started/os_setup.html#anaconda-installation 를 참고하기 바란다. 아나콘다를 설치해놓고, tensorflow 환경(environment)를 정의한 후에, 주피터 노트북을 설치하면 된다. http://stackoverflow.com/questions/37061089/trouble-with-tensorflow-in-jupyter-notebook 참고


Tensorflow 환경을 만든 후에,

$ source activate tensorflow

를 실행해서 텐서 플로우 환경으로 전환한후, 아래와 같이 ipython 을 설치한후에, 주피터 (jupyter) 노트북을 설치하면 된다.

(tensorflow) username$ conda install ipython
(tensorflow) username$ pip install jupyter #(use pip3 for python3)


아나콘다 기반의 텐서플로우 환경 설정은 나중에 시간이 될때 다른 글을 통해서 다시 설명하도록 하겠다.

텐서플로우의 자료형

텐서플로우는 뉴럴네트워크에 최적화되어 있는 개발 프레임웍이기 때문에, 그 자료형과, 실행 방식이 약간 일반적인 프로그래밍 방식과 상의하다. 그래서 삽질을 많이 했다.


상수형 (Constant)

상수형은 말 그대로 상수를 저장하는 데이타 형이다.

  • tf.constant(value, dtype=None, shape=None, name='Const', verify_shape=False)

와 같은 형태로 정의 된다. 각 정의되는 내용을 보면

  • value : 상수의 값이다.

  • dtype : 상수의 데이타형이다. tf.float32와 같이 실수,정수등의 데이타 타입을 정의한다.

  • shape : 행렬의 차원을 정의한다. shape=[3,3]으로 정의해주면, 이 상수는 3x3 행렬을 저장하게 된다.

  • name : name은 이 상수의 이름을 정의한다. name에 대해서는 나중에 좀 더 자세하게 설명하도록 하겠다.

간단한 예제를 하나 보자.

a,b,c 상수에, 각각 5,10,2 의 값을 넣은 후에, d=a*b+c 를 계산해서 계산 결과 d를 출력하려고 한다.

import tensorflow as tf


a = tf.constant([5],dtype=tf.float32)

b = tf.constant([10],dtype=tf.float32)

c = tf.constant([2],dtype=tf.float32)


d = a*b+c


print d

그런데, 막상 실행해보면, a*b+c의 값이 아니라 다음과 같이 Tensor… 라는 문자열이 출력된다.


Tensor("add_8:0", shape=(1,), dtype=float32)

그래프와 세션의 개념

먼저 그래프와 세션이라는 개념을 이해해야 텐서플로우의 프로그래밍 모델을 이해할 수 있다.

위의 d=a*b+c 에서 d 역시 계산을 수행하는 것이 아니라 다음과 같이 a*b+c 그래프를 정의하는 것이다.


실제로 값을 뽑아내려면, 이 정의된 그래프에 a,b,c 값을 넣어서 실행해야 하는데, 세션 (Session)을 생성하여,  그래프를 실행해야 한다. 세션은 그래프를 인자로 받아서 실행을 해주는 일종의 러너(Runner)라고 생각하면 된다.


자 그러면 위의 코드를 수정해보자


import tensorflow as tf


a = tf.constant([5],dtype=tf.float32)

b = tf.constant([10],dtype=tf.float32)

c = tf.constant([2],dtype=tf.float32)


d = a*b+c


sess = tf.Session()

result = sess.run(d)

print result



tf.Session()을 통하여 세션을 생성하고, 이 세션에 그래프 d를 실행하도록 sess.run(d)를 실행한다

이 그래프의 실행결과는 리턴값으로 result에 저장이 되고, 출력을 해보면 다음과 같이 정상적으로 52라는 값이 나오는 것을 볼 수 있다.


플레이스 홀더 (Placeholder)

자아 이제 상수의 개념을 알았으면, 이제는 플레이스 홀더에 대해서 알아보자.

y = x * 2 를 그래프를 통해서 실행한다고 하자. 입력값으로는 1,2,3,4,5를 넣고, 출력은 2,4,6,8,10을 기대한다고 하자. 이렇게 여러 입력값을 그래프에서 넣는 경우는 머신러닝에서 y=W*x + b 와 같은 그래프가 있다고 할 때, x는 학습을 위한 데이타가 된다.

즉 지금 살펴보고자 하는 데이타 타입은 학습을 위한 학습용 데이타를 위한 데이타 타입이다.


y=x*2를 정의하면 내부적으로 다음과 같은 그래프가 된다.


그러면, x에는 값을 1,2,3,4,5를 넣어서 결과값을 그래프를 통해서 계산해 내야한다. 개념적으로 보면 다음과 같다.



이렇게 학습용 데이타를 담는 그릇을 플레이스홀더(placeholder)라고 한다.

플레이스홀더에 대해서 알아보면, 플레이스 홀더의 위의 그래프에서 x 즉 입력값을 저장하는 일종의 통(버킷)이다.

tf.placeholder(dtype,shape,name)

으로 정의된다.

플레이스 홀더 정의에 사용되는 변수들을 보면

  • dtype : 플레이스홀더에 저장되는 데이타형이다. tf.float32와 같이 실수,정수등의 데이타 타입을 정의한다.

  • shape : 행렬의 차원을 정의한다. shapre=[3,3]으로 정의해주면, 이 플레이스홀더는 3x3 행렬을 저장하게 된다.

  • name : name은 이 플레이스 홀더의 이름을 정의한다. name에 대해서는 나중에 좀 더 자세하게 설명하도록 하겠다.


그러면 이 x에 학습용 데이타를 어떻게 넣을 것인가? 이를 피딩(feeding)이라고 한다.

다음 예제를 보자


import tensorflow as tf


input_data = [1,2,3,4,5]

x = tf.placeholder(dtype=tf.float32)

y = x * 2


sess = tf.Session()

result = sess.run(y,feed_dict={x:input_data})


print result


처음 input_data=[1,2,3,4,5]으로 정의하고

다음으로 x=tf.placeholder(dtype=tf.float32) 를 이용하여, x를 float32 데이타형을 가지는 플레이스 홀더로 정의하다. shape은 편의상 생략하였다.

그리고 y=x * 2 로 그래프를 정의하였다.


세션이 실행될때, x라는 통에 값을 하나씩 집어 넣는데, (앞에서도 말했듯이 이를 피딩이라고 한다.)

sess.run(y,feed_dict={x:input_data}) 와 같이 세션을 통해서 그래프를 실행할 때, feed_dict 변수를 이용해서 플레이스홀더 x에, input_data를 피드하면, 세션에 의해서 그래프가 실행되면서 x는 feed_dict에 의해서 정해진 피드 데이타 [1,2,3,4,5]를 하나씩 읽어서 실행한다.


변수형 (Variable)

마지막 데이타형은 변수형으로,

y=W*x+b 라는 학습용 가설이 있을때, x가 입력데이타 였다면, W와 b는 학습을 통해서 구해야 하는 값이 된다.  이를 변수(Variable)이라고 하는데, 변수형은 Variable 형의 객체로 생성이 된다.


  • tf.Variable.__init__(initial_value=None, trainable=True, collections=None, validate_shape=True, caching_device=None, name=None, variable_def=None, dtype=None, expected_shape=None, import_scope=None)


변수형에 값을 넣는 것은 다음과 같이 한다.


var = tf.Variable([1,2,3,4,5], dtype=tf.float32)


자 그러면 값을 넣어보고 코드를 실행해보자


import tensorflow as tf


input_data = [1,2,3,4,5]

x = tf.placeholder(dtype=tf.float32)

W = tf.Variable([2],dtype=tf.float32)

y = W*x


sess = tf.Session()

result = sess.run(y,feed_dict={x:input_data})


print result


우리가 기대하는 결과는 다음과 같다. y=W*x와 같은 그래프를 가지고,


x는 [1,2,3,4,5] 값을 피딩하면서, 변수 W에 지정된 2를 곱해서 결과를 내기를 바란다.

그렇지만 코드를 실행해보면 다음과 같이 에러가 출력되는 것을 확인할 수 있다.



이유는 텐서플로우에서 변수형은 그래프를 실행하기 전에 초기화를 해줘야 그 값이 변수에 지정이 된다.


세션을 초기화 하는 순간 변수 W에 그 값이 지정되는데, 초기화를 하는 방법은 다음과 같이 변수들을 global_variables_initializer() 를 이용해서 초기화 한후, 초기화된 결과를 세션에 전달해 줘야 한다.


init = tf.global_variables_initializer()

sess.run(init)


그러면 초기화를 추가한 코드를 보자


import tensorflow as tf


input_data = [1,2,3,4,5]

x = tf.placeholder(dtype=tf.float32)

W = tf.Variable([2],dtype=tf.float32)

y = W*x


sess = tf.Session()

init = tf.global_variables_initializer()

sess.run(init)

result = sess.run(y,feed_dict={x:input_data})


print result


초기화를 수행한 후, 코드를 수행해보면 다음과 같이 우리가 기대했던 결과가 출력됨을 확인할 수 있다.



텐서플로우를 처음 시작할때, Optimizer나 모델등에 대해 이해하는 것도 중요하지만, “데이타를 가지고 학습을 시켜서 적정한 값을 찾는다" 라는 머신러닝 학습 모델의 특성상, 모델을 그래프로 정의하고, 세션을 만들어서 그래프를 실행하고, 세션이 실행될때 그래프에 동적으로 값을 넣어가면서 (피딩) 실행한 다는 기본 개념을 잘 이해해야, 텐서플로우 프로그래밍을 제대로 시작할 수 있다.


Fluentd + Bigquery + Jupyter를 이용한 초간단 BI 구축하기


조대협

얼마전에 빅데이타의 전문가로 유명한 김형준님이 "Presto + Zeppelin을 이용한 초간단 BI 구축 사례"라는 발표 자료를 보았다. http://www.slideshare.net/babokim/presto-zeppelin-bi 오픈 소스 기술들을 조합하여, 초간단하게 빅데이타 분석 플랫폼을 만든 사례 인데, 상당히 실용적이기도 하고, 좋은 조합인것 같아서, 마침 구글 빅쿼리에 대한 자료를 정리하던중 비슷한 시나리오로 BI 대쉬 보드를 만들어보았다.

Fluentd를 이용해서 실시간으로 데이타를 수집하고, 이를 빅쿼리에 저장한 다음에 iPython nodebook (aka Jupyter)로 대쉬보드를 만드는 예제이다. 일부 제품에 대한 지식이 없었음에도 불구하고 실제 설정은 대략 2시간 정도 걸렸다.


아래 이제 예제는 정상적으로 작동 하지 않습니다. 트위터에서 JSON 스키마를 변경했는데, 거기에 맞는 빅쿼리 JSON 스키마를 구하기가 어렵네요. (만들자니 귀찮고). 참고로만 사용하세요



Fluentd 설치

예제는 Google Cloud에서 Ubuntu Linux 14.x VM에서 Fluentd를 이용하여 Twitter에서 데이타를 읽은 후, 빅쿼리에 데이타를 로딩하는 시나리오이다.

VM 생성

Fluentd를 설치할 VM을 생성해보자. 구글 클라우드 콘솔에서 아래 그림과 같이 VM을 생성할때, “Identity and API access” 부분에  “Allow full access to all Cloud APIs”를 선택한다. 이를 선택해서 이 VM이 모든 구글 클라우드 API에 대한 접근 권한 (BigQuery 포함)을 가지도록 한다.


tdagent 설치

생성한 VM에 fluentd의 로그 수집 에이전트인 tdagent를 설치한다.

tdagent는 OS나, 또는 같은 OS라도 OS 버전별로 설치 방법이 다르기 때문에, 버전별 설치 방법은 http://www.fluentd.org를 참고하기 바란다.

여기서는 Ubuntu 14.x를 기준으로 진행을 하였다.

다음 명령어를 실행하면 tdagent가 설치된다.

% curl -L https://toolbelt.treasuredata.com/sh/install-ubuntu-trusty-td-agent2.sh | sh

설치한 후 에이전트를 실행해서 확인해보자. 다음 명령으로 agent를 실행한 후에,

% sudo /etc/init.d/td-agent restart

실행이 끝난 후에 다음 명령으로 설치를 확인한다.

% sudo /etc/init.d/td-agent status


참고 (tdagent 관련 명령어)

tdagent 기동 - $sudo /etc/init.d/td-agent start
tdagent 정지 - $sudo /etc/init.d/td-agent stop
tdagent 재시작 - $sudo /etc/init.d/td-agent restart
tdagent 상태확인 - $sudo /etc/init.d/td-agent status




트위터 Input 설정하기

tdagent 에이전트 설치가 끝났으면 fluentd를 이용해서 트위터 피드를 읽어드리도록 해보자.

트위터 API 키 받기

트위터 피드는 트위터에서 제공하는 OPEN API를 통해서 읽어드린다. 그래서 이 OPEN API에 접근하기 위해서는 OPEN API키가 필요하다.

OPEN API 키는 https://apps.twitter.com/ 에 접속하고 Create New App 메뉴를 이용하면 새로운 앱을 등록할 수 있고, 여기에 Fluentd 앱을 정의해서 정보를 넣어주고 Key and secrect을 생성해주면 다음과 같이 키가 생성된 것을 웹에서 확인할 수 있다.


여기서 필요한 키값은 Consumer Key, Consumer Secret, Access Token, Access Token Secret 4가지가 필요하다.

트위터 플러그인 설치하기

API 접근을 위한 API Key를 모두 얻었으면 이제 fluentd에서 트위터 피드를 읽기 위한 트위터 플러그인을 설치해보자.

트위터 API는 libssl에 대한 의존성이 있기 때문에, libssl를 먼저 설치한다.

%sudo apt-get install build-essential libssl-dev

다음 트위터 플러그인이 사용하는 eventmachine 플러그인과, 트위터 플러그인을 설치한다.

% sudo td-agent-gem install eventmachine

% sudo td-agent-gem install fluent-plugin-twitter

설정하기

플러그인 설치가 끝났으면 설정을 해보자. 설정 파일은 /etc/td-agent/td-agent.conf 에 있다.

이 파일을 다음과 같이 편집하자.


<source>

 type twitter

 consumer_key        {앞서 트위터 콘솔에서 받은 Consumer Key}

 consumer_secret     {앞서 트위터 콘솔에서 받은 Consumer  secret}

 oauth_token         {앞서 트위터 콘솔에서 받은 Access token}

 oauth_token_secret {앞서 트위터 콘솔에서 받은 Access token secret}

 tag                 input.twitter.sampling  # Required

 timeline            sampling                # Required (tracking or sampling or location or userstream)

 keyword             galaxy,game        # 검색어

 output_format       nest                   # Optional (nest or flat or simple[default])

</source>

<match input.twitter.sampling>

 type stdout

</match>


이 설정 파일은 keyword에 등록된 “galaxy”와 “game” 이라는 키워드를 찾아서, 읽어드린후 <match input.twitter.sampling> 에 의해서, 읽어드린 내용을 stdout으로 출력해주는 설정이다.

테스트

설정이 끝났으면 확인을 해보자

% sudo /etc/init.d/td-agent restart

명령어를 수행하여, td-agent를 리스타트 해서 새로운 config 파일이 반영되도록 하고

% tail -f /var/log/td-agent/td-agent.log          

를 통해서 stdout으로 올라오는 로그를 확인하자. 제대로 데이타가 수집되는 것을 확인했으면 다음 명령어를 이용해서, td-agent를 정지 시키자.

% sudo /etc/init.d/td-agent stop


빅쿼리로 저장하기

twitter로 부터 피드를 읽어드리는 플러그인이 정상적으로 작동함을 확인하였으면, 이번에는 읽어드린 데이타를 빅쿼리로 저장해보자.

빅쿼리 플러그인 설치 및 테이블 생성

빅쿼리로 데이타를 쓰기 위해서 빅쿼리 플러그인을 설치한다.

% sudo td-agent-gem install fluent-plugin-bigquery


다음으로 빅쿼리 프로젝트에서 트위터 데이타를 저장할 데이타셋과 테이블을 생성한다.

데이타 셋 이름은 편의상 “twitter”라고 하고, 테이블은 “ timeline”이라고 하고 생성을 하겠다.

테이블의 스키마는 트위터 피드에 대한 데이타 구조를 빅쿼리 스키마로 만들어놓은 스키마가 이미 https://gist.github.com/Salinger/ef39b81ad2c48516b596

에 있기 때문에, 이 스키마 파일을 읽어서 빅쿼리 콘솔에서 아래 그림과 같이 Schema 부분에 Copy & Paste를 해서 붙이면 테이블이 생성된다.


설정하기

테이블이 생성이 되었으면 fluentd 설정 파일을 수정하여 트위터 피드를 이 테이블에 저장하도록 설정한다.


<source>
 type twitter
   consumer_key        {앞서 트위터 콘솔에서 받은 Consumer Key}

 consumer_secret     {앞서 트위터 콘솔에서 받은 Consumer  secret}

 oauth_token         {앞서 트위터 콘솔에서 받은 Access token}

 oauth_token_secret {앞서 트위터 콘솔에서 받은 Access token secret}

 tag                 input.twitter.sampling  # Required
 timeline            sampling                # Required (tracking or sampling or location or userstream)
 keyword             hillary,clinton,donald,trump
 output_format       nest                    # Optional (nest or flat or simple[default])
</source>

<match input.twitter.sampling>
 type copy
<store>
  type bigquery
  buffer_type file
  buffer_path /var/log/td-agent/buffer/twi.*.buf
  method insert

  auth_method compute_engine
  project useful-hour-138023
  dataset twitter
  table timeline

  flush_interval 1
  buffer_chunk_limit 1000000
  buffer_queue_limit 5000
  flush_interval 1
  try_flush_interval 0.05
  num_threads 4
  queue_chunk_flush_interval 0.01

  time_format %s
  time_field log_time
  schema_path /home/terrycho/bq_tweet.json
  log_level error
</store>
</match>


기존 설정 파일에서 <match input.twitter.sampling> 부분을 빅쿼리로 변경하였다. <store>에서 type을 bigquery로 변경하였다.

중요한 필드들을 살펴보면

  • buffer_type, buffer_path : fluentd는 트위터에서 읽어드리는 데이타를 건건이 bigquery에 저장하는게 아니라 일정 단위로 모아서 bigquery에 저장한다. 그래서 buffer를 사용하는데, buffer를 파일을 이용하고, 이 파일의 위치를 지정해주었다.

  • auth_method, project,dataset,table : 데이타를 저장한 bigquery의 project,dataset,table 명을 정한다. 그리고 auth_method를 통해서 인증 방법을 설정하는데, 일반적으로는 service account에 대한 json 파일을 사용하는데, 여기서는 구글 클라우드내에 VM을 생성하였고, 앞에서 VM 생성시에 Bigquery에 대한 접근 권한을 이미 주었기 때문에, 인증 방식을 compute_engine으로 설정하면 된다.

  • flush_interval 은 어떤 주기로 버퍼된 데이타를 bigquery로 저장할것인지를 정한다. 여기서는 1초 단위로 저장하도록 하였다.

  • 그리고 중요한것중 하나가 schema_path 인데, 저장하고자 하는 bigquery 테이블의 스키마이다. 앞에서 테이블 생성에서 사용한 https://gist.github.com/Salinger/ef39b81ad2c48516b596 에서 다운 받았던 *.json으로 정의된 스키마 파일의 경로를 지정해주면 된다.

실행하기

모든 설정이 끝났으면

%sudo /etc/init.d/td-agent restart

명령을 이용해서 tdagent를 재기동하자.

그리고 빅쿼리 콘솔에서 “select count(*) from 테이블명” 명령을 사용하면 아래와 같이 카운트 수가 매번 올라가면서 데이타가 저장되는 것을 확인할 수 있다.


Datalab으로 대쉬보드 만들기

datalab은 오픈소스 iPython note의 구글 클라우드 버전이다. 자동으로 구글 클라우드 내의 앱앤진 내에 설치해주고, 구글 클라우드의 빅데이타 인프라 (빅쿼리등)과 손쉽게 연동되며, 구글 차트를 내장하고 있어서 그래프도 손쉽게 그려줄 수 있다.


데이타랩 준비하기

데이타랩을 사용하기 위해서는 https://datalab.cloud.google.com/ 에 접속하고, 로그인을 하면 다음과 같이 프로젝트를 선택하는 화면이 나온다.


만약에 아직 데이타랩을 설치 하지 않았으면 가운데 Deploy 버튼만 활성화가 된다. Deploy 버튼을 누르면 자동으로 데이타랩이 설치된다. 설치가 끝나면 Start 버튼이 활성화 된다. Start 버튼을 누르면 데이타 랩으로 들어갈 수 있다.

새로운 노트 만들기

다음은 데이타랩의 초기화면이다.


우리는 여기서, 새로운 노트를 만들어서 앞서 빅쿼리로 읽어드린 데이타를  lang(언어)별로 그룹핑을 해서 카운트하는 쿼리를 실행하고, 그 결과를 그래프로 만들것이다.

위의 초기화면에서 “+Notebook” 버튼을 눌려서 새로운 노트북을 만들어보자


노트화면이 로딩되었으면 상단의 메뉴를 보자


+Add code와, +Add Markdown 버튼을 볼 수 있는데,  Add Code는 파이썬이나 SQL과 같은 프로그래밍 언어를 정의하고 실행할 수 있는 공간이고, +Add Markdown은 일반적인 텍스트나 이미지를 통해서 간단한 글을 쓸 수 있는 공간을 만들어준다.

이렇게 코드써가면서 직접 실행해보고 그 결과를 확인하면서 그에 대한 내용을 설명하는 내용을 Markdown으로 작성하는 것과 같이 마치 노트에 계산을 해나가는 것처럼 써 나가기 때문에 이런 류의 프로그램을 노트북이라고 한다. (유사한 프로그램으로는 zeppelin 등이 있다 https://zeppelin.apache.org/)

쿼리 실행하기

그러면 Add code를 통해서 코드 섹션을 추가하고 SQL 문장을 추가해보자. 다음은 빅쿼리 트위터 테이블에서 lang 별로 그룹핑을 해서 카운트를 하는 SQL 문장이다.


이 문장을 실행하려면 노트북 상단의 “Run” 버튼을 누르면 된다.

다음과 같이 결과가 쿼리 바로 아래에 출력되는 것을 볼 수 있다.




그래프 그리기

다음으로 결과로 그래프를 그려보자

다음과 같이 두개의 코드 블럭을 추가하자


첫번째 코드 블럭에는 SQL 문장을 수행하는데 이때 --module twitter라고 정의를 해주면 결과가 twitter라는 모듈에 저장이 된다.

두번째 코드 블럭은 그래프를 그리기 위해서 chart 명령어를 이용하고 차트 타입은 pie로, 그래프의 x,y 축은 lang과, lang_count로 지정하고, 데이타 소스는  --date를 이용해서 앞의 쿼리 결과를 저장한 twitter로 지정한다.

다음으로 Run 버튼을 이용해서 쿼리를 수행해보면 다음과 같은 결과 화면을 얻을 수 있다.





지금까지 간략하게 Fluentd를 통해서 데이타를 수집하고 빅쿼리에 저장한 후, 데이타랩을 통해서 분석 및 리포팅을 하는 간단한 시나리오를 살펴보았다. fluentd나 데이타랩에 대한 사전적인 지식이 없었는데, 필자의 경우 이를 만드는데 대략 2시간의 시간이 소요되었다. 2시간의 시간으로 수 PB급의 빅데이타를 수집할 수 있고 분석할 수 있는 시스템을 구축할 수 있었다. 예전 같으면 하둡과 스팍 인스톨과 몇시간이 걸렸는데, 요즘 드는 생각은 빅데이타에 대한 접근 장벽이 많이 무너졌다고나 할까.

참고 자료


동적 연결 알고리즘 (Dynamic Connectivity)


조대협 (http://bcho.tistory.com)


요즘 알고리즘이 대세라 기초를 다지는 차원에서 다시 알고리즘을 보고 있는데, 오늘은 동적 연결 알고리즘에 대해서 공부한 내용을 간략하게 정리해본다.


동적 연결 알고리즘은 노드끼리 연결이 되어 있는지를 찾는 알고리즘이다.

각 도시간에 연결이 되어 있는지, SNS에서 친구끼리 서로 연결이 되어 있는지와 같은 연결성만을 판단한다.


예전에 이러한 문제를 그래프 형태의 자료구조를 이용해서 풀려고 했는데, 여기서 문제의 핵심은 노드 A와 B가 연결되었는지만 판단하면 된다. 즉 자료 구조상에서, 어떤 노드가 어떤 노드와 인접해 있는지등은 표시할 필요가 없다. 아래 그림을 보자, 아래 그림에서 1과 5는 연결이 되어 있다. (1->2->5 경로) 4,2는 연결되어 있지 않다.



이 처럼 노드간의 연결 여부만 판단 하는 것이 동적 연결 알고리즘 (Dynamic Connectivity)문제이다.

 

QuickUnionFind


이를 구현하는데 몇가지 알고리즘이 있는데, 가장 간단한 알고리즘으로는 QucikUnionFind라는 알고리즘이다.

QuickUionFind 알고리즘은 아래와 같이 일차원 배열을 이용해서 구현하는데, 배열의 Index가 노드명, 그 값이 ID이다. ID가 같은 노드들은 서로 연결되어 있는 것으로 취급 한다.



이런식으로 된다. 값이 9,8,7 3의 컴포넌트로 나뉘어 짐을 알 수 있다.

이 알고리즘의 경우 구현은 쉽지만, 매번 연결을 추가할때 마다 ID값을 스캔해서, 같은 컴포넌트에 있는 노드의 ID값을 매번 변경해야 하기 때문에 성능이 매우 낮다. O(N^2)의 복잡도를 갔는다.


Tree (QuickFindUF)


그래서 이를 보완하는 알고리즘으로는 Tree를 사용하는 방법이 있는데, Tree의 값에는 이 노드와 연결된 상위 노드의 인덱스 값을 가지게 한다.

이때 주의해야 하는 것이, Tree에서 표현되는 상하위 노드의 연결 내용은 중요하지 않다. 실제로 그림에서 1,3 노드간에는 연결은 없지만, 필요하다면 Tree에서는 1,3 노드를 연결해서 표현하되, 1,3이 하나의 트리에만 있으면 된다.

헷갈릴 수 있는 부분이, 각 노드간의 연결을 온전히 저장하는 것이 아니라, 컴포넌트에 어떤 노드가 있는지를 저장하는 것이기 때문에, 각 노드 간의 연결성에 초점을 맞출 필요가 없다. 



위의 그래프 저장 결과는 배열에서 다음과 같다.

 



여기서 노드가 연결되어 있는지를 찾으려면, root 노드가 같은지를 비교하면 된다.

즉, 노드 9,5가 연결되어 있는지 찾으려면 9 --> 5 까지의 경로를 트래버스 하는 것이 아니라, 9와 5의 root 노드(1)을 찾아서 동일한지를 비교하면 된다.


WeightTree

위의 트리 기반의 알고리즘의 경우, Depth가 비약적으로 길어질 수 있는 문제가 있는데, 이를 해결하기 위해서는 WeightTree라는 알고리즘을 사용할 수 있다. 이 알고리즘은 두개의 트리를 병합할때, 항상 작은 트리를 큰 트리의 루트에 붙이게 함으로써 전체 트리의 Depth를 줄이는 방법이다.


아마 이 알고리즘을 보지 않았더라면, Dynamic Connectivity 문제도 Node *  기반의 자료 구조를 만들어서 DFS(깊이 우선 검색)으로 해결하려고 하지 않았을까? (성능은 무지 안나오겠지)


참고자료 및 소스

http://algs4.cs.princeton.edu/15uf/