InfluxDB, Redis, MQTT 정리

InfluxDB

개요

InfluxDB는 시계열 데이터(time series data)를 효율적으로 저장하고 처리하기 위해 설계된 오픈 소스 시계열 데이터베이스입니다. IoT 센서 데이터, 애플리케이션 모니터링 지표, 실시간 분석 등에 주로 사용됩니다.

주요 특징

• 시계열 데이터 최적화: 시간에 따른 데이터 저장 및 검색에 최적화
• 높은 수집 및 쿼리 성능: 초당 수백만 개의 데이터 포인트 처리 가능
• 자동 데이터 보존 정책: 오래된 데이터 자동 삭제 기능
• 내장 HTTP API: RESTful API를 통한 쉬운 데이터 접근
• Flux/InfluxQL: 강력한 쿼리 언어 제공
• 내장 시각화 도구(Chronograf): 데이터 시각화 및 관리 도구

핵심 개념

1. 데이터 구조

InfluxDB의 데이터 모델은 다음과 같은 요소로 구성됩니다:

구성 요소 설명 예시

측정(Measurement)	시계열 데이터의 컨테이너(일반 DB의 테이블과 유사)	cpu_usage, temperature
태그(Tag)	메타데이터를 저장하는 키-값 쌍 (인덱싱됨)	host=server01, region=us-west
필드(Field)	실제 측정값을 저장하는 키-값 쌍 (인덱싱되지 않음)	value=0.64, temperature=22.5
타임스탬프(Timestamp)	각 데이터 포인트의 시간 정보	1635739200000000000

2. 라인 프로토콜

데이터를 InfluxDB에 쓰는 기본 형식:

<measurement>[,<tag_key>=<tag_value>] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]

예시:

cpu_usage,host=server01,region=us-west user=0.64,system=0.13,idle=86.23 1635739200000000000

기본 사용법

설치 (Ubuntu 예시)

# 저장소 추가
wget -qO- https://repos.influxdata.com/influxdb.key | sudo apt-key add -
source /etc/lsb-release
echo "deb https://repos.influxdata.com/${DISTRIB_ID,,} ${DISTRIB_CODENAME} stable" | sudo tee /etc/apt/sources.list.d/influxdb.list

# 설치
sudo apt-get update && sudo apt-get install influxdb

# 서비스 시작
sudo systemctl start influxdb

데이터베이스 생성 및 사용

-- 데이터베이스 생성
CREATE DATABASE mydb

-- 데이터베이스 사용
USE mydb

데이터 쓰기

# HTTP API 사용
curl -i -XPOST 'http://localhost:8086/write?db=mydb' --data-binary 'cpu_usage,host=server01 value=0.64'

# CLI 사용
influx -database 'mydb' -execute "INSERT cpu_usage,host=server01 value=0.64"

데이터 조회

-- 기본 쿼리
SELECT * FROM cpu_usage WHERE time > now() - 1h

-- 집계 함수 사용
SELECT mean(value) FROM cpu_usage WHERE host='server01' GROUP BY time(10m)

-- 모든 태그 값 조회
SHOW TAG VALUES FROM cpu_usage WITH KEY = "host"

연속 쿼리 (Continuous Queries)

자동으로 주기적인 데이터 다운샘플링 수행

CREATE CONTINUOUS QUERY "cq_30m" ON "mydb"
BEGIN
  SELECT mean(value) AS mean_value
  INTO "cpu_usage_30m"
  FROM "cpu_usage"
  GROUP BY time(30m), *
END

보존 정책 (Retention Policies)

데이터 자동 삭제 정책 설정

-- 30일 보존 정책 생성
CREATE RETENTION POLICY "thirty_days" ON "mydb" DURATION 30d REPLICATION 1 DEFAULT

InfluxDB 2.0+ 변경사항

InfluxDB 2.0부터 몇 가지 주요 개념이 변경되었습니다:

• 버킷(Bucket): 데이터베이스와 보존 정책을 대체
• Flux 쿼리 언어: InfluxQL을 대체하는 더 강력한 쿼리 언어
• 조직(Organization): 멀티 테넌시 지원을 위한 새로운 개념
• 토큰 기반 인증: 사용자 인증을 위한 토큰 시스템

사용 사례

• IoT 센서 데이터 저장
• 애플리케이션 성능 모니터링
• 서버 및 인프라 모니터링
• 실시간 분석 및 대시보드
• 금융 시계열 데이터 분석

성능 최적화 팁

1. 태그와 필드를 적절히 구분하여 사용
2. 고카디널리티(high cardinality) 태그 사용 자제
3. 적절한 보존 정책 설정
4. 샤딩 및 클러스터링 고려
5. 메모리 및 캐시 설정 최적화

Redis

개요

Redis(Remote Dictionary Server)는 인메모리 데이터 구조 저장소로, 키-값 저장소, 캐시, 메시지 브로커 등으로 사용됩니다. 매우 빠른 읽기/쓰기 성능과 다양한 데이터 구조를 지원하여 실시간 애플리케이션에 적합합니다.

주요 특징

• 인메모리 저장: 모든 데이터를 메모리에 저장하여 빠른 접근 속도
• 다양한 데이터 구조: 문자열, 해시, 리스트, 세트, 정렬된 세트 등 지원
• 영속성 옵션: RDB 스냅샷, AOF 로그를 통한 데이터 영속화
• 복제 및 클러스터링: 고가용성 및 확장성 지원
• 트랜잭션: 원자적 작업 지원
• Pub/Sub 기능: 메시지 브로커 기능 제공
• Lua 스크립팅: 서버 측 스크립트 실행 지원

핵심 개념

1. 데이터 구조

데이터 구조 설명 주요 명령어

String	텍스트 또는 바이너리 데이터(최대 512MB)	SET, GET, INCR
Hash	필드-값 쌍의 컬렉션(객체 저장에 적합)	HSET, HGET, HMGET
List	문자열의 순서가 있는 컬렉션(큐나 스택으로 사용)	LPUSH, RPUSH, LPOP
Set	순서가 없는 고유한 문자열의 컬렉션	SADD, SMEMBERS, SINTER
Sorted Set	스코어로 정렬된 고유한 문자열 컬렉션	ZADD, ZRANGE, ZRANK
Bitmap	비트 연산을 지원하는 특수 문자열	SETBIT, GETBIT, BITCOUNT
HyperLogLog	고유 항목 수 추정에 사용되는 확률적 자료구조	PFADD, PFCOUNT
Stream	로그나 시계열 데이터에 적합한 새로운 자료구조	XADD, XREAD, XRANGE

2. 데이터 만료 (Expiration)

모든 키에 대해 만료 시간을 설정할 수 있습니다

SET key value
EXPIRE key 60  # 60초 후 만료

 

또는

SETEX key 60 value  # 값 설정과 만료 시간 설정을 한 번에

3. 영속성 옵션

• RDB (Redis Database): 주기적으로 데이터 스냅샷을 디스크에 저장
• AOF (Append-Only File): 모든 쓰기 작업을 로그 파일에 기록
• 혼합 모드: RDB와 AOF 함께 사용

기본 사용법

설치 (Ubuntu 예시)

sudo apt update
sudo apt install redis-server

# 서비스 시작
sudo systemctl start redis-server

기본 명령어

문자열 작업

SET username "johndoe"
GET username
INCR counter
DECR counter
MSET key1 "value1" key2 "value2"
MGET key1 key2

 

해시 작업

HSET user:1000 username "johndoe" email "john@example.com"
HGET user:1000 username
HMGET user:1000 username email
HGETALL user:1000

 

리스트 작업

LPUSH tasks "task1"
RPUSH tasks "task2"
LRANGE tasks 0 -1  # 모든 항목 조회
LPOP tasks  # 왼쪽에서 항목 제거 및 반환
RPOP tasks  # 오른쪽에서 항목 제거 및 반환

 

세트 작업

SADD tags "redis" "database" "nosql"
SMEMBERS tags
SISMEMBER tags "redis"
SINTER set1 set2  # 교집합
SUNION set1 set2  # 합집합

 

정렬된 세트 작업

ZADD scores 90 "Alice"
ZADD scores 85 "Bob"
ZADD scores 95 "Carol"
ZRANGE scores 0 -1  # 낮은 점수부터 모든 항목
ZREVRANGE scores 0 -1  # 높은 점수부터 모든 항목
ZRANGEBYSCORE scores 85 95  # 점수로 범위 지정

Pub/Sub 패턴

발행자-구독자 메시징 패턴 구현

구독자

SUBSCRIBE channel1

 

발행자

PUBLISH channel1 "Hello, Redis!"

트랜잭션

여러 명령을 원자적으로 실행

MULTI
SET key1 "value1"
SET key2 "value2" 
EXEC

Redis 클라이언트 라이브러리

Node.js (ioredis)

const Redis = require("ioredis");
const redis = new Redis();

async function example() {
  await redis.set("key", "value");
  const value = await redis.get("key");
  console.log(value);
}

 

Python (redis-py)

import redis
r = redis.Redis(host='localhost', port=6379, db=0)

r.set('key', 'value')
value = r.get('key')
print(value)

 

Java (Jedis)

Jedis jedis = new Jedis("localhost");
jedis.set("key", "value");
String value = jedis.get("key");
System.out.println(value);
jedis.close();

고급 기능

1. Redis Sentinel

마스터-슬레이브 구성의 고가용성 보장

• 자동 장애 조치
• 모니터링
• 알림

2. Redis Cluster

수평적 확장을 위한 분산 Redis 구현

• 자동 샤딩
• 클러스터 노드 간 데이터 분산
• 고가용성

3. Lua 스크립팅

EVAL "return redis.call('SET', KEYS[1], ARGV[1])" 1 mykey myvalue

사용 사례

• 캐싱 시스템
• 세션 저장소
• 실시간 분석
• 리더보드 및 카운터
• 메시지 큐
• 실시간 채팅 애플리케이션
• 지리 공간 인덱싱

성능 최적화 팁

1. 메모리 사용량 모니터링 (INFO memory)
2. 키 이름 최적화 (짧은 키 이름 사용)
3. 파이프라이닝을 통한 네트워크 왕복 최소화
4. 적절한 데이터 구조 선택
5. 대량 쓰기 작업에는 MULTI/EXEC 사용

 

하기와 같이 따라한다면 에러가 뜨지 않을 것이다.

 

MQTT

개요

MQTT(Message Queuing Telemetry Transport)는 경량 메시지 프로토콜로, IoT(사물인터넷) 기기 간의 통신에 널리 사용됩니다. 제한된 대역폭과 리소스 환경에서도 효율적으로 작동하도록 설계되었습니다.

주요 특징

• 발행/구독(Publish/Subscribe) 패턴: 중앙 브로커를 통한 메시지 교환
• 경량화: 최소한의 프로토콜 오버헤드 (2바이트부터 가능)
• QoS(Quality of Service) 수준: 메시지 전달 보장 수준 제공
• 영구 세션: 재연결 시 구독 상태 유지 가능
• Last Will and Testament: 클라이언트 비정상 종료 시 자동 알림
• 보안: TLS/SSL 및 사용자 인증 지원

핵심 개념

1. MQTT 구성 요소

• 브로커(Broker): 메시지를 수신하고 구독자에게 배포하는 중앙 서버
• 클라이언트(Client): 브로커에 연결하여 메시지를 발행하거나 구독하는 디바이스
• 토픽(Topic): 메시지를 분류하는 계층적 구조의 문자열
• 메시지(Message): 토픽을 통해 교환되는 페이로드

2. 토픽 구조

토픽은 계층적 구조를 가진 문자열로, /로 구분됩니다

home/livingroom/temperature
building/floor1/room3/humidity

와일드카드 사용:

• +: 단일 수준 와일드카드 (예: home/+/temperature)
• #: 다중 수준 와일드카드 (예: home/#)

3. QoS(Quality of Service) 수준

QoS 레벨 설명 사용 사례

0 (최대 한 번)	메시지를 한 번만 전송, 전달 보장 없음	데이터 손실이 허용되는 센서 데이터
1 (최소 한 번)	메시지가 최소 한 번 전달됨, 중복 가능	중요한 알림, 로그 데이터
2 (정확히 한 번)	메시지가 정확히 한 번 전달됨	결제, 제어 명령 등 중요 데이터

기본 사용법

브로커 설치 (Mosquitto 예시)

# Ubuntu에서 설치
sudo apt update
sudo apt install mosquitto mosquitto-clients

# 서비스 시작
sudo systemctl start mosquitto

기본 구성 (mosquitto.conf)

# 기본 리스너 설정
listener 1883
allow_anonymous true

# 로그 설정
log_dest file /var/log/mosquitto/mosquitto.log

명령줄로 테스트

구독하기

mosquitto_sub -h localhost -t "test/topic" -v

 

발행하기

mosquitto_pub -h localhost -t "test/topic" -m "Hello, MQTT!"

클라이언트 라이브러리 사용

Node.js (MQTT.js)

const mqtt = require('mqtt');
const client = mqtt.connect('mqtt://localhost:1883');

// 구독
client.on('connect', () => {
  client.subscribe('test/topic');
});

// 메시지 수신
client.on('message', (topic, message) => {
  console.log(`${topic}: ${message.toString()}`);
});

// 메시지 발행
client.publish('test/topic', 'Hello from Node.js!');

 

Python (Paho-MQTT)

import paho.mqtt.client as mqtt

# 콜백 정의
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("test/topic")

def on_message(client, userdata, msg):
    print(f"{msg.topic}: {msg.payload.decode()}")

# 클라이언트 설정
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# 브로커 연결 및 루프 시작
client.connect("localhost", 1883, 60)
client.loop_start()

# 메시지 발행
client.publish("test/topic", "Hello from Python!")

 

ESP8266/ESP32 (Arduino)

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

const char* ssid = "YourWiFi";
const char* password = "YourPassword";
const char* mqtt_server = "broker.example.com";

WiFiClient espClient;
PubSubClient client(espClient);

void setup_wifi() {
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
  }
}

void callback(char* topic, byte* payload, unsigned int length) {
  // 메시지 처리
  String message;
  for (int i = 0; i < length; i++) {
    message += (char)payload[i];
  }
  Serial.println(message);
}

void reconnect() {
  while (!client.connected()) {
    if (client.connect("ESP8266Client")) {
      client.subscribe("test/topic");
    } else {
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback);
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();
  
  // 주기적으로 메시지 발행
  client.publish("test/sensor", "22.5");
  delay(2000);
}

고급 기능

1. 영구 세션 (Persistent Sessions)

클라이언트가 재연결할 때 구독 상태를 유지:

// Clean session을 false로 설정
const client = mqtt.connect('mqtt://localhost', {
  clientId: 'clientId',
  clean: false
});

2. Last Will and Testament (LWT)

클라이언트가 비정상적으로 연결 종료될 때 전송될 메시지:

const client = mqtt.connect('mqtt://localhost', {
  will: {
    topic: 'device/status',
    payload: 'offline',
    qos: 1,
    retain: true
  }
});

3. 유지(Retained) 메시지

토픽에 마지막으로 저장된 메시지를 새 구독자에게 전송:

mosquitto_pub -h localhost -t "office/temperature" -m "23.5" -r

4. MQTT 보안

• TLS/SSL 암호화: 통신 암호화
• 사용자 인증: 사용자 이름/비밀번호 인증
• ACL(Access Control List): 클라이언트별 토픽 접근 권한 설정

Mosquitto 보안 설정 예시

# mosquitto.conf
listener 8883
cafile /etc/mosquitto/ca_certificates/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key

# 사용자 인증 활성화
allow_anonymous false
password_file /etc/mosquitto/passwd

사용 사례

• IoT 센서 데이터 수집
• 스마트 홈 자동화
• 실시간 모니터링 시스템
• 자산 추적
• 푸시 알림
• 채팅 애플리케이션
• 분산 제어 시스템

모범 사례

1. 토픽 설계: 의미 있고 계층적인 토픽 구조 설계
2. 적절한 QoS 레벨 선택: 사용 사례에 맞는 QoS 설정
3. 보안 적용: TLS/SSL 및 인증 구현
4. 클라이언트 ID: 고유한 클라이언트 ID 사용
5. Keep-Alive 타이머: 적절한 Keep-Alive 시간 설정
6. 메시지 크기: 작은 메시지 크기 유지 (대형 데이터는 분할 전송)
7. 에러 처리 및 재연결 로직: 강건한 클라이언트 구현

MQTT vs 다른 프로토콜 비교

프로토콜 특징 사용 사례

MQTT	경량, Pub/Sub 모델, QoS	IoT, 저대역폭 환경
HTTP/REST	요청/응답 모델, 상태 없음	웹 API, 단순 통합
WebSocket	양방향 통신, 이벤트 기반	실시간 웹 애플리케이션
CoAP	REST 유사, UDP 기반, 경량	제약 있는 IoT 기기
AMQP	고급 큐잉, 트랜잭션	엔터프라이즈 메시징