IoT 센서 스트림과 Isolation Forest로 이상 탐지 운영하기

연속으로 들어오는 센서 이벤트를 이상 신호로 바꾸는 경로

연결된 센서가 늘수록 데이터는 개별 레코드가 아니라 끊임없는 흐름으로 쌓인다. 이 흐름을 제때 분석하지 못하면 장애나 품질 문제를 사전에 포착하기 어렵다.

IoT 스트림 처리는 센서 이벤트를 초·밀리초 단위로 수집하고 변환·분석하는 방식이다. 마이크로배치나 실제 이벤트 기반 모델로 구성하며, 메시지 브로커(MQTT/Kafka), 윈도우와 워터마크, 상태 관리(State), 정확히 한 번(Exactly-once) 처리 의미론이 핵심이 된다.

이상 탐지는 정상 패턴과 통계적으로 유의하게 다른 이벤트를 찾아내는 기법이다. 사전 레이블이 부족한 환경에서는 비지도 학습이 주로 쓰인다. Isolation Forest(IF)는 샘플을 무작위 분할 트리로 고립시키는 데 필요한 분할 깊이로 이상도를 계산하는 비지도 모델이다. 고차원·비정규 분포에 강건하고 학습·추론 속도가 좋다.

스트림에서 IF를 사용할 때는 롤링 윈도우 재학습이나 마이크로배치 재적합 전략을 둔다. 순증형 온라인 학습은 지원하지 않는다는 점을 운영 설계에 반영해야 한다.

수집부터 알림까지 이어지는 데이터 파이프라인

데이터 경로는 엣지에서 시작해 게이트웨이, 브로커(MQTT/Kafka), 스트림 엔진(Flink/Spark/Kafka Streams), 피처 엔지니어링, IF 추론, 알림·저장으로 이어진다. Avro/Protobuf와 스키마 레지스트리로 진화형 스키마를 관리하고, 파싱 또는 검증에 실패한 이벤트는 DLQ(Dead Letter Queue)에 격리한다.

시간 처리는 이벤트 타임을 기준으로 잡는다. 슬라이딩·텀블링 윈도우와 워터마크를 적용해 지연 및 순서 뒤바뀜(out-of-order)을 처리한다. Kafka 트랜잭션과 Flink 체크포인트를 결합하면 E2E exactly-once 경로를 구성할 수 있으며, 재처리 시에는 Idempotent sink로 중복을 막아야 한다.

피처는 윈도우 집계에서 얻는 평균·표준편차·이동 IQR, FFT 성분 같은 스펙트럼 피처, 도메인 규칙 기반 파생 변수로 구성한다. 결측과 스파이크는 Z-score 또는 Median Absolute Deviation으로 완화하고, 센서 캘리브레이션 변동에는 표준화로 대응한다.

IF의 하이퍼파라미터는 n_estimators(100~~300), max_samples(0.5~~1.0), contamination(사내 기준·분위수 기반), max_features(0.5~~1.0)를 기준으로 잡을 수 있다. 스트림에서는 롤링 윈도우 재학습 주기를 설정하고(예: 5~~15분), Warm start로 모델을 안정화한다. 점수 임계값은 최근 점수 분포의 동적 분위수로 보정한다.

관측 대상에는 처리 지연, 워터마크 지연, 드롭률, 경보율(Precision/Recall), 오탐율이 포함된다. 전송 구간은 TLS/MQTT over TLS로 보호하고, Kafka ACL/SASL과 PII 최소 수집·마스킹을 적용한다. 엣지 단에서는 secure boot와 TPM으로 위변조를 방지한다.

운영 환경별 적용 모습

제조 설비 진동 이상은 진동 센서에서 Kafka로 이벤트를 보내고, Flink의 1s 슬라이딩 윈도우에서 FFT+RMS를 계산한 뒤 IF 점수로 경보를 생성하고 CMMS 티켓으로 연결한다. 미세 베어링 손상을 조기에 포착해 계획 정비로 전환하고 다운타임을 줄이는 경로다.

스마트 빌딩에서는 전력·온도 센서 데이터를 Kafka Streams에서 시계열 트렌드와 피크 감지에 사용한다. IF가 비정상 소비를 탐지하면 BAS 제어 신호로 이어져 불필요 부하 차단과 피크 요금 절감을 지원한다.

콜드체인 물류에서는 온습도·문열림 이벤트를 Spark Structured Streaming으로 처리한다. 구간별 편차와 체류 시간 피처를 IF에 전달하고, 결과에 따라 경로 재배차나 알림을 수행해 콜드체인 위반을 실시간으로 차단하고 폐기율을 낮춘다.

지연과 오탐을 함께 다루는 운영 기준

이 구조는 평균 탐지 지연(Latency) 500ms~~5s 수준을 달성하고 배치 대비 10~~100배 단축할 수 있다. 동적 임계와 충분한 피처링을 전제로 오탐율은 20~40% 감소하며, 설비 다운타임은 15% 이상 감소한다.

도메인 규칙과 모델을 병행하면 현장의 수용성도 높아진다. 데이터 기반 선제 대응 체계를 갖추고 운영 신뢰성과 투명성을 높이는 효과가 있다.

입력 단계에서는 센서에서 엣지 게이트웨이(MQTT QoS1/2, TLS)를 거쳐 Kafka에 인입한다. 압축은 snappy/zstd를 사용하고 파티션 키는 device_id로 둔다. 이어서 Avro/Protobuf 스키마와 범위·단위·결측값을 검증한다.

처리 단계에서는 스트림 엔진이 이벤트 타임을 정렬하고 워터마크(예: 5s)를 설정해 슬라이딩 윈도우 피처를 만든다. IF가 점수를 산출하고 동적 분위수 기반 임계값으로 판단하며, 재학습 스케줄러는 롤링 윈도우로 모델을 갱신한다. Kafka Source, Flink 체크포인트, 트랜잭셔널 Kafka Sink를 연결해 exactly-once 경로를 구성한다.

출력은 경보 토픽·웹훅·CMMS, 시각화 대시보드, 데이터 레이크·TSDB로 보낼 수 있다. 파싱 실패는 DLQ로, 피처 생성 실패는 Fallback 규칙 경보로, 모델 미가용은 규칙 기반 임계 동작으로 처리한다. 지연이 심화되거나 워터마크를 초과한 이벤트는 별도 경로에 보관한 뒤 오프라인에서 재중앙화한다.

⤢✕

Kafka 마이크로배치에서 IF를 재학습하는 예시

전제는 Python 3.10+, scikit-learn 1.4+, confluent-kafka, numpy, pandas다. 레이블은 없으며 롤링 윈도우 재학습 전략을 사용하고, 재학습 주기는 N 배치로 가정한다.

scikit-learn IsolationForest는 온라인 학습을 지원하지 않아 부분 업데이트가 불가능하다. 아래 코드는 롤링 윈도우 재적합으로 이를 다루는 개념 증명 수준이며, 정확히 한 번 처리는 Flink 같은 엔진에서 구현하는 편이 적합하다.

# pip install scikit-learn==1.4.2 confluent-kafka numpy pandas joblib
import json, time
from collections import deque
import numpy as np
from confluent_kafka import Consumer
from sklearn.ensemble import IsolationForest
from joblib import dump, load

KAFKA_BOOTSTRAP = "localhost:9092"
TOPIC = "sensor.raw"
GROUP = "if-stream-consumer"
WINDOW_SIZE = 20000          # 롤링 학습 윈도우 크기
RETRAIN_EVERY = 3000         # N 메시지마다 재학습
CONTAMINATION = 0.02
SEED = 42

def build_consumer():
    return Consumer({
        "bootstrap.servers": KAFKA_BOOTSTRAP,
        "group.id": GROUP,
        "auto.offset.reset": "latest",
        "enable.auto.commit": True
    })

def extract_features(msg_dict):
    # 예시: 숫자형 필드 5개 사용
    f = [
        msg_dict.get("vibration_rms", 0.0),
        msg_dict.get("temp_c", 0.0),
        msg_dict.get("current_a", 0.0),
        msg_dict.get("speed_rpm", 0.0),
        msg_dict.get("humidity", 0.0),
    ]
    return np.array(f, dtype=np.float32)

buf = deque(maxlen=WINDOW_SIZE)
scores_buf = deque(maxlen=5000)
model = IsolationForest(
    n_estimators=200, max_samples=0.7, contamination=CONTAMINATION,
    max_features=1.0, bootstrap=False, random_state=SEED, warm_start=True
)

def dynamic_threshold():
    if len(scores_buf) < 500:
        # 초기 안정화 단계: 이론적 컷(대략적) 사용
        return np.quantile(list(scores_buf) if scores_buf else [0.0], 0.02)
    # 최근 점수 하위 분위수(예: 2%)를 임계로 사용
    return np.quantile(scores_buf, CONTAMINATION)

def retrain_if_needed(counter):
    if len(buf) < 5000:  # 최소 학습 샘플
        return
    if counter % RETRAIN_EVERY == 0:
        X = np.stack(buf, axis=0)
        model.set_params(n_estimators=model.n_estimators)  # warm_start 유지
        model.fit(X)

def main():
    c = build_consumer()
    c.subscribe([TOPIC])
    counter = 0
    try:
        while True:
            msg = c.poll(1.0)
            if msg is None or msg.error():
                continue
            try:
                payload = json.loads(msg.value().decode("utf-8"))
                x = extract_features(payload)
            except Exception:
                # 파싱 실패 시 DLQ로 전송하는 로직 권장. 여기서는 무시.
                continue

            buf.append(x)
            counter += 1
            retrain_if_needed(counter)

            if len(buf) >= 5000:
                # 추론 및 동적 임계값 적용
                score = model.score_samples(x.reshape(1, -1))[0]  # 높을수록 정상
                scores_buf.append(score)
                thr = dynamic_threshold()
                is_anomaly = score < thr
                if is_anomaly:
                    # TODO: 알림 토픽/웹훅 전송
                    print(f"[ALERT] device={payload.get('device_id')} score={score:.4f} thr={thr:.4f}")
    except KeyboardInterrupt:
        pass
    finally:
        c.close()

if __name__ == "__main__":
    main()

피처 스케일링은 학습 윈도우에 맞춰 표준화 또는 로버스트 스케일러를 재적합한다. 라인이나 장비별로 모델을 나누면 도메인 편차에 대응할 수 있다. 안전 임계값 같은 하드 룰을 먼저 적용하고 IF를 정밀 경보의 후속 조치에 쓰는 룰+IF 앙상블도 운영 선택지다.

스트림 엔진을 고를 때 보는 차이

항목	Flink	Spark Structured Streaming	Kafka Streams
성능	저지연, 이벤트기반, 백프레셔 강함	마이크로배치/Continuous 모드, 중저지연	초저지연, JVM 내 임베디드
확장성	대규모 상태·체크포인트 강점	대규모 배치/스트리밍 통합	애플리케이션 수평 확장 용이
일관성	E2E exactly-once 성숙	exactly-once(일부 소스/싱크 제약)	트랜잭션 기반 exactly-once
안정성	풍부한 상태 복구/세이브포인트	탄탄한 커뮤니티/운영 경험	간결하나 복잡 파이프라인 한계
운영 편의	세밀 설정 필요, 러닝커브 높음	기존 Spark 생태계 활용 용이	배포 간단, JVM 서비스와 친화

정확히 한 번 처리와 지연은 맞바꾸는 요소다. 체크포인트 간격을 촘촘히 하면 중복·유실 위험은 줄지만 처리 지연과 I/O 부담은 늘어난다. SLA를 기준으로 균형점을 정한다.

재학습 주기가 짧으면 드리프트 대응력은 높아지지만 CPU·메모리와 팟 스케일이 증가한다. 윈도우 크기와 주기는 그리드 탐색 대상으로 다룬다. 고정 contamination은 계절성과 부하 변동에 취약하므로 동적 분위수와 시계열 분해(추세/계절성 분리)를 함께 적용한다.

경보 품질을 판단하는 지표

실시간성은 처리 지연(p50/p95), 워터마크 지연, 유효 처리율(EPS)로 측정한다. 정확도는 Precision/Recall/F1, 오탐율(FPR), 탐지 소요시간(MTTD)을 본다. 안정성은 재처리율, DLQ 비율, 체크포인트 성공률로 확인한다.

초기 도입은 규칙과 IF를 결합한 하이브리드 방식으로 시작할 수 있다. 이벤트 타임·워터마크·상태 관리로 처리 신뢰도를 확보하고, 롤링 재학습과 동적 임계로 드리프트에 대응한다. 라이브러리와 프레임워크는 버전별 호환성 차이가 있으므로 최신 정보를 확인한다.