Data Analysis : Prologue - EMQ + Kafka + openTSDB + Grafana

Data Analysis : Prologue - EMQ + Kafka + openTSDB + Grafana

사실 이 포스팅의 제목을 어떻게 정해야 할지 고민이 많았다. 일단 클러스터 설치는 마쳤으니 당연히 이전 포스팅의
카테고리에서는 벗어나야 할 것이고, 실제 데이터를 다루기 시작하였으니 데이터 분석의 시작은 시작인데…
수집되는 데이터 자체가 딱히 분석할만한 데이터는 아니고…암튼 조금은 모호한 지점에 있는 작업이 되어버렸다.
더군다나 본격적인 데이터 분석은 아직 한참 더 공부한 이후 시작될 터인데…

하지만 역시나 데이터를 다루기 시작했으니 데이터 관련 제목을 붙여야 하겠다고 결정을 했다. Prologue라는 부제와
함께. 이후 The Beginning을 시작하게 되겠지만 아마도 시간이 조금 걸리지 않을까 싶다. 오늘은 가볍게나마
이전에 클러스터 구성 내용을 정리할 때 누락된 openTSDB와 Grafana 설치에 대한 내용과 EMQ -> Kafka ->
openTSDB -> Grafana로 이어지는 데이터 수집/저장/시각화 과정을 정리해보도록 하겠다.

시나리오

직전 포스팅(아두이노 온습도계 제작)에서도 언급했듯이 좀 더 그럴싸한 데이터를 모아 분석을 해보고 싶었지만
막상 작업을 시작하고 보니 마땅히 모을 수 있는 데이터가 없었다. 그렇다고 데이터를 찾아 다닐수만은 없기에
우선 쉽게 접근할 수 있는 것으로부터 시작해보기로 했다. 시나리오는 다음과 같다.

1. 아두이노로 제작된 온습도 및 먼지 센서를 통해 온도,습도,먼지 농도,시간 데이터 수집
2. MQTT 프로토콜을 통해 EMQ 서버로 데이터 전송
3. EMQ 서버로 publishing 된 데이터를 Kafka에서 subscribing
4. Kafka에 들어온 데이터를 openTSDB에 저장
5. openTSDB에 저장된 데이터를 Grafana로 시각화
6. Spark를 이용하여 데이터 분석 (이상 감지)

EMQ를 통해 M2M 통신을 하는 상황이거나 데이터 누락을 어느 정도 허용하는 작업이라면 굳이 Kafka가 필요하진
않을 것이다. 하지만 일단 데이터를 수집하고 분석하는 작업을 전제로 하였기에 누락되는 데이터를 Kafka에서 1차로
완충하도록 구성하였다. 최종 분석단계에 해당하는 6번 Spark를 이용한 데이터 분석은 아직 좀 더 학습이 진행되어야
할 부분이다.

openTSDB 설치 및 실행

최근 시계열 데이터를 처리하는 NoSQL로는 influxDB가 대세인듯하다. 다만 openTSDB가 HBase를 기반으로
작동한다는 한 가지 이유만으로 나는 openTSDB를 선택하였다. 설치 및 연동 과정에서 후회를 많이 했지만 결국
불굴의 의지로 제대로 연동 시키고야 말았다…ㅠ.ㅠ influxDB가 대세라는 것은 구글링만 해봐도 충분하다. 이놈에
openTSDB는 검색을 해도 설치 외에는 쓸만한 정보가 없더라…-.-

간단하게 openTSDB의 개요를 말하자면 TSD라는 데몬을 통해 외부의 시계열 데이터를 HBase에 생성된 테이블에
저장하는 구조이다.

설치 시 유의사항

현재 최신 릴리즈는 2.3.0 이며 pre-release가 2.4.0 RC2까지 나와있다. 아래 설명할 kafka와의 연동을
위한 opentsdb-rpc-kafka라는 plugin을 사용하기 위해서는 반드시 2.4.0 RC2 버전을 설치해야 한다.
나는 이 사실을 모르고 2일간 삽질을 했다…ㅠ.ㅠ 

설치는 역시 간단하다. 압축 파일을 다운로드 받아 적당한 디렉토리(나는 당근 /opt)에서 압축을 풀고 압축을 푼
디렉토리 내에 있는 build.sh를 실행시키는 것만으로 설치는 끝이다. 

다만 앞서 말한 것처럼 openTSDB는 HBase 기반으로 작동을 하기 때문에 HBase가 이미 설치 되어있다는 것을 
전제로 한다. 나는 이미 이전에 클러스터를 구성하였기에 바로 설치에 들어갔다.

설치가 끝나면 build.sh 파일이 있는 그 경로에 build라는 경로가 추가되고 그 안에 openTSDB 실행 파일 등이
위치하게 된다. 설치가 끝나면 가장 먼저 테이블을 생성해주어야 한다. 테이블 생성 스크립트는 build/src 아래에
있다. 아래와 같이 실행해준다.

$ cd /opt/opentsdb/build
$ env COMPRESSION=NONE HBASE_HOME=/opt/hbase ./src/create_table.sh

이렇게 실행하면 ‘tsdb’, ‘tsdb-uid’, ‘tsdb-tree’, ‘tsdb-meta’ 이렇게 4개의 테이블이 HBase 상에 생성된다.

그리고 마지막으로 설정을 한다. 설정파일은 최초에 압축을 푼 디렉토리 밑에 있는 src 디렉토리에 있다.
나의 경우에는 /opt/opentsdb/src가 그 경로이며 파일 이름은 opentsdb.conf이다. 이 파일을 build로
복사한다. 즉, /opt/opentsdb/build로 복사를 하는 것이다. 수정해야 할 설정 내용은 간단하다.

# The TCP port TSD should use for communications
# *** REQUIRED ***
tsd.network.port = 4242
...
# The location of static files for the HTTP GUI interface.
# *** REQUIRED ***
tsd.http.staticroot = /opt/opentsdb/build/staticroot

# Where TSD should write it's cache files to
# *** REQUIRED ***
tsd.http.cachedir = /tmp/opentsdb
...
# Whether or not to automatically create UIDs for new metric types, default
# is False
tsd.core.auto_create_metrics = true

여기까지 모두 마쳤으면 /opt/opentsdb/build아래 있는 tsdb를 다음과 같이 실행해준다.

$ /opt/opentsdb/build/tsdb tsd

실행한 콘솔창에 로그가 올라가는 것이 보일 것이고 localhost:4242로 접속을 하면 아래 이미지와 같은 콘솔
화면을 볼 수 있다.

센서 데이터의 포맷 변경

기존에 아두이노 온습도계에서 EMQ 서버로 전송하던 데이터는 포맷을 좀 바꾸어야 한다. openTSDB에 저장하기
위한 데이터 포맷은 JSON 형식으로 모두 3가지 유형이 있다.

• Metric
• Aggregate
• Histogram

자세한 설명은 생략하고…-.- 나는 가장 기본적인 Metric 형태로 데이터를 저장하기로 했다. Metric 형태로 저장할
경우 JSON 포맷은 다음과 같다(여러건을 한번에 보내려면 아래 형식을 배열 안에 넣으면 된다).

{
    "metric": "sys.cpu.nice",
    "timestamp": 1346846400,
    "value": 18,
    "tags": {
       "host": "web01",
       "dc": "lga"
    }
}

metric은 전체 측정 단위를, timestamp는 이름 그대로 long 타입의 타임스탬프를 value는 측정 값을, tags는
세부 분류를 위한 키-값 쌍을 추가하면 된다. 나의 예로 좀더 자세하게 살펴보자.

[
   {
      "type": "Metric",
      "metric": "mqtt.home.pcroom",
      "timestamp": 1346846400,
      "value": 22.5,
      "tags": {
         "type": "temperature",
         "loc": "pcroom"
      }
   },
   {
      "type": "Metric",
      "metric": "mqtt.home.pcroom",
      "timestamp": 1346846400,
      "value": 18.2,
      "tags": {
         "type": "humidity",
         "loc": "pcroom"
      }
   },
   {
      "type": "Metric",
      "metric": "mqtt.home.pcroom",
      "timestamp": 1346846400,
      "value": 22.3,
      "tags": {
         "type": "dust",
         "loc": "pcroom"
      }
  }
]

위 내용을 살펴보면 우선 같은 시점에 측정된 온도/습도/먼지의 측정값을 배열로 넘기고 있다. 이 데이터를
만드는 것은 아두이노에서 처리하며 관련 내용은 아래 링크의 소스코드 부분을 참조하면 된다.

아두이노를 이용한 온도/습도/먼지 측정기

최상위에 있는 type 값은 openTSDB 자체를 위한 것이 아니라 opentsdb-rpc-kafka plugin에서 필요한
것이다. 3가지 유형 중 하나를 지정하기 위한 속성이다.

metric의 값은 임으로 정하면 되는 것으로 나의 경우 그 의미는 mqtt를 이용하여 home에 있는 컴퓨터방
pcroom의 측정값이라는 뜻이다. 

timestamp와 value는 달리 설명할 부분이 없고 tags에 보면 type에 각각 temperature, humidity, dust
라는 값이 들어있는데 의미 그대로 온도와 습도와 먼지를 뜻한다.

이렇게 3개의 값을 하나의 배열에 포함시켜 보내는 것이다. tags에서 loc는 큰 의미는 없다. 필요에 따라 추가적인
값을 지정하면 된다.

Kafka와 openTSDB 연동

openTSDB 자체에 대한 자료 혹은 openTSDB와 Grafana 연동에 대한 자료는 그럭저럭 찾을 수 있었는데
Kafka와 openTSDB와의 연동에 대한 자료는 거의 찾을 수가 없었다. 구글링을 하면 가장 상위에 검색되는
내용이 2개인데 하나는 opentsdb-rpc-kafka라는 openTSDB의 plugin이고 다른 하나는 
kafka-connect-opentsdb라는 Kafka connector이다. 뭔가 사용하기에는 connector쪽이 쉬워보였으나
openTSDB의 HTTP API를 이용한다는 점이 맘에 들지 않았다. 그래서 opentsdb-rpc-kafka를 선택하게
되었고 그렇게 고난의 길은 시작되었다…ㅠ.ㅠ

일단 opentsdb-rpc-kafka는 Github에서 소스를 내려받은 후 빌드를 하여 jar 파일을 만들어야 한다.
소스는 Maven 프로젝트로 되어있어 이클립스에서 빌드하거나 또는 프로젝트 디렉토리 아래에서 man package
명령으로 빌드하면 된다.

플러그인 소스 경로는 다음과 같다.

https://github.com/OpenTSDB/opentsdb-rpc-kafka

이렇게 빌드된 jar 라이브러리를 /opt/opentsdb/build 아래에 plugins 디렉토리를 만들어 그 안에 넣는다.
plugin 라이브러리의 최종 위치는 다음과 같다. 그리고 이 plugin의 경로는 반드시 classpath에 추가시켜주어야 한다.

$ /opt/opentsdb/build/plugins/opentsdb-rpc-kafka-2.3.2-SNAPSHOT.jar

그리고 opentsdb.conf 설정 파일에 plugin과 관련된 내용을 추가해주어야 하는데 아래와 같은 내용을 설정 파일의
제일 하단에 추가해주면 된다. 물론 각 설정 값은 각자의 환경에 맞게 입력해야 한다. 고정이라고 주석을 달아놓은
설정 외에는 각자 환경에 맞춰 수정을 하자.

# --------- PLUGINS ---------------------------------
tsd.core.plugin_path = /opt/opentsdb/build/plugins
## 고정
tsd.core.storage_exception_handler.enable = true
## 고정
tsd.core.storage_exception_handler.plugin = net.opentsdb.tsd.KafkaStorageExceptionHandler
## 고정
tsd.rpc.plugins = net.opentsdb.tsd.KafkaRpcPlugin
KafkaRpcPlugin.kafka.zookeeper.connect = rpi1:2181,rpi2:2181,rpi3:2181
KafkaRpcPlugin.kafka.metadata.broker.list = rpi1:9092,rpi2:9092,rpi3:9092
KafkaRpcPlugin.groups = mqtt
KafkaRpcPlugin.mqtt.topics = mqtt-kafka
## 고정
KafkaRpcPlugin.mqtt.consumerType = raw
## 고정
KafkaRpcPlugin.mqtt.deserializer = net.opentsdb.data.deserializers.JSONDeserializer
#KafkaRpcPlugin.mqtt.rate = 5
KafkaRpcPlugin.mqtt.threads = 3
## 고정
KafkaRpcPlugin.seh.topic.default = seh

앞서 말한 바와 같이 이 플러그인을 사용하기 위해서는 openTSDB 2.4.0 RC2 버전을 설치해야 한다.
한동안 자바 개발을 등한시 했더니 아주 사소한 것을 파악하지 못해 삽질을 따따블로 했다. 분명 빌드는 잘 되는데
plugin 설치하고 openTSDB를 기동하면 쌩뚱맞게 속성에 접근하지 못한다느니, 메소드가 존재하지 않는다느니
하는 메시지가 나오면서 plugin이 제대로 작동하지 않는 것이다.

나중에 확인해보니 소스에 포함된 openTSDB 라이브러리는 2.4.0 버전이고 내가 설치한 openTSDB는 2.3.0
이었다. 전체적인 패키지 구조는 같지만 클래스 내에 변화가 있었기 때문에 속성이나 메소드 사용에 문제가
있었던 것이다. 확인 후 부랴부랴 2.4.0 RC2 버전으로 다시 설치를 하였다.

이제 다시 한 번 openTSDB를 실행해보자(이미 실행중이라면 실행 중인 콘솔에서 ctrl+c를 눌러 종료시키자).
드디어 Kafka에서 데이터를 쭈~욱 쭉 뽑아오는 모습을 볼 수 있을 것이다.

 

이 데이터는 이전에 클러스터 관련 포스팅 중 EMQ와 Kafka의 연동을 통해 가져온 데이터들이다. 관련 내용은 아래 
링크에서 확인할 수 있다.

Cluster : The Beginning - Apache Kafka와 EMQ 연동

주의 사항

EMQ에서 Kafka로 전송된 데이터의 포맷은 형태가 다르다. 실제로 openTSDB에 저장하기 위한 내용은
payload 부분에 들어있는데 이 payload의 값은 Base64 인코딩이 되어 저장된다. kafka와 EMQ 연동을
위해 사용한 connector에서 그렇게 처리를 하고 있다. 따라서 opentsdb-rpc-kafka plugin에서 데이터를
제대로 처리하기 위해서는 Kafka에서 받아온 값에서 payload의 값만을 추출한 뒤 이 값을 Base64 디코딩
하여 전달해야 한다. 

이 부분을 처리해야 하는 소스는 net.opentsdb.tsd.KafkaRpcPluginThread.java의 run 함수이다.

이렇게 모든 과정이 끝나면 앞서 localhost:4242로 접근했던 웹 콘솔을 통해 쿼리가 가능하다. 그런데 어쩐 일인지
나의 경우 오류가 발생하면서 데이터 조회가 되지 않았다. 콘솔창에 찍히는 URL이 조금 이상해 보이긴 한데 그것이
원인인지는 잘 모르겠다. 결국 openTSDB 자체 웹 화면에서 데이터를 조회하는 것은 성공하지 못했다.

Grafana 설치 및 설정

적어도 설치에 있어서는 지금껏 설치한 모든 시스템 중 Grafana가 가장 친절하였다.
운영체제별로 다운로드 버튼이 있고 이 버튼을 누르면 해당 운영체제에 설치하는 방법이 나온다.
나는 현재 Mac mini에 설치를 하고 있기 때문에 Mac에 대한 설치 방법을 제공해 주는 것이 얼마나
반가웠는지 모른다…ㅠ.ㅠ 설치는 매우 간단하다.

brew update 
brew install grafana

이렇게 설치를 하고나면 다음과 같은 내용이 콘솔 화면에 표시된다.

To have launchd start grafana now and restart at login:
  brew services start grafana
Or, if you don't want/need a background service you can just run:
  grafana-server —config=/usr/local/etc/grafana/grafana.ini —homepath /usr/local/share/grafana cfg:default.paths.logs=/usr/local/var/log/grafana cfg:default.paths.data=/usr/local/var/lib/grafana cfg:default.paths.plugins=/usr/local/var/lib/grafana/plugins

친절하게 Grafana의 실행 방법을 알려주는 것이다. 나는 그냥 간단하게 brew를 통해 실행을 하였다.
설정파일은 상당히 긴 편인데 따로 변경해주어야 할 것은 아무것도 없다. 웹 화면의 포트를 바꾸는 정도?

openTSDB와의 연동

설정을 바꾸지 않았다면 localhost:3000으로 Grafana 웹 콘솔에 접근할 수 있다. 처음 접속하면 계정을 물어보는데
기본 계정은 admin / admin이다. 접속을 해서 가장 먼저 수행해야 하는 작업은 Datasource를 연결하는 것이다.
Grafana는 다양한 DB와 연동 가능한데 type 항목에서 select box를 클릭하면 다음과 같이 목록이 표시된다.

나머지 설정들은 자료도 많고 하니 참고해서 입력하면 되는데 나같은 경우 HTTP settings에서 Access를 Direct로
설정하니 데이터 소스에 연결이 되지 않았다. Access를 proxy로 설정하고 HTTP Auth를 With Credentials로
설정하니 비로소 데이터소스 연결에 성공했다는 메시지가 표시되었다. 이 과정에서도 왜 연결이 안되는지 Grafana의
소스까지 까뒤집고 난리를 치면서 상당한 시간을 보냈다…ㅠ.ㅠ 전체적인 설정은 다음과 같다(OpenTSDB settings
의 Version은 현재 2.3까지만 선택 가능한데 2.4.0 RC2를 설치한 경우에도 2.3을 선택하면 된다).

Datasource를 성공적으로 연결했다면 이제 대시보드를 만들면 된다. 좌측 메인 메뉴에서 Dashbord에 마우스를
올리면 나타나는 서브 메뉴에서 + New를 선택하여 새로운 대시보드를 만드는데 처음 나타나는 화면은 대시보드에
표현할 유형을 선택하는 화면이다. 나머지는 차차 알아보고 나는 우선 맨 앞의 Graph를 선택했다.

Graph를 선택하면 아래 이미지와 같이 비어있는 Graph panel이 덩그러니 하나 나온다. 여기서 상단의 Panel Title을
클릭하면 몇가지 메뉴가 나오는데 Edit를 선택하자

Edit를 선택하면 Graph panel 하단으로 Graph 조건을 입력하는 화면이 나온다. 역시 다른 항목들은 차차
알아보기로 하고 중요한 몇가지만 살펴보자.

먼저 General 항목으로 가서 Graph 이름을 지정해준다. 그러면 Graph 화면 상단에 있던 Panel Title이라는
문구가 지정해준 이름으로 바뀐다. 일단 TEST라고 입력해 보았다.

다음 가장 중요한 Metric 설정이다. 역시 많은 항목이 있지만 꼭 필요한 2가지만 설정해보자.
먼저 Metric 값을 입력한다. 기억하시겠지만 openTSDB로 넘기는 JSON 문자열에 metric이라는 키가 있는데
바로 그 값을 넣어주면 된다. 나는 mqtt.home.pcroom이었다. Metric 값만 입력해도 벌써 챠트가 그려지는데
나의 경우 하나의 Metric에 온도/습도/먼지 3가지 데이터가 들어있었다. 

이 시점에 나타나는 그래프는 설정화면의 Metric 옆에 보이는 Aggregator가 sum으로 되어있기 때문에 
온도 + 습도 + 먼지의 값으로 그려진 그래프다. 한마디로 쓰레기다…-.- 물론 이런 값이 필요한 경우도 분명 있을 
것이다. 각각의 데이터로 그래프를 그리기 위해서는 tags에 지정한 구분값을 추가로 입력해주어야 한다. 

아래 그림과 같이 내가 tags에 지정한 type키와 그에 대한 값인 temperature를 입력해주었다. 입력 후 우측에 
있는 add tag 버튼을 눌러야 반영된다.

이제 온도에 대한 데이터만 Graph에 표시된다.
그렇다면 하나의 Graph에 여러 데이터를 표시하려면 어떻게 하면 될까?
Edit 영역 하단에 보면 Add Query라는 파란색 버튼이 있다. 이 버튼을 클릭하면 동일한 입력 폼이 하나 더 생기며
이렇게 새로 생긴 입력 폼에 tags를 구분해서 입력해주면 된다.

최종 화면은 요렇게 보인다.

정리

이거 은근히 내용이 길어져버렸다. 달리 생각하면 이 작업에 그만큼 많은 시행착오와 노력이 들어갔다는 말이 되겠지만
최근에는 시계열 데이터도 Elasticsearch와 Kibana를 이용하여 처리하는 것이 대세인 것 같아 내가 이러려고
openTSDB를 선택했던가 하는 자괴감이 든다는 철지난 개그가 절로 나오게 되었다…ㅠ.ㅠ 아무래도 본격적으로
데이터 처리를 하게 되면 바로 이 Elasticsearch + Kibana 조합으로 가지 않을까 싶다.

그래도 클러스터 구성에서부터, 작은 데이터를 가지고 클러스터의 일부분만 사용하는 작업이긴 하지만 실제 데이터를
다루어보았다는 점에서 상당한 만족감을 느낀다.

원래 최종 목표는 이렇게 수집한 데이터를 인공지능으로 분석하는 것이었다. 그리고 그 내용에는 트위터의 데이터를
수집하여 RNN 분석을 하는 것도 있었는데 앞으로의 계획이 너무 거창해져서 일단 그 부분은 보류를 해야 할 것 같다.

앞으로의 계획은 이렇다.

1. 아두이노와 메카넘휠을 이용한 차량형 로봇 제작, 수집 대상은 모터 회전수, 초음파 센서를 이용한 장애물과의 거리 데이터, 충돌 센서를 이용하여 장애물과의 충돌이 발생했을 때의 false 정보, 바퀴의 둘레와 모터 회전수를 조합한 이동 거리 정보, 기타 부가 정보로 카메라를 이용한 영상 정보 등이다.
2. 정보의 활용은 2가지이다. DQN을 이용한 강화학습을 통해 낮은 수준의 자율주행 구현이 그 하나이고 gazebo simulator를 이용한 Digital Twin을 구현하는 것이 다른 한가지다.

무식하면 용감하다고…그래도 그간 조금씩 습득해온 지식으로는 뭔가 어렵지 않게 될 것 같기도 한데 현실은 
어떨지…-.-

이제 긴 여정을 위해 잠시 학습하는 시간을 갖고 이후 블로그 포스팅은 주로 학습 내용을 정리하는 수준이 될 것 같다.
그럼 다음에…