Spark 스트리밍 파이프라인에 ONNX 추론을 끼워 넣는 법

추천 모델 추론이 수백 밀리초 SLA를 넘기면 사용자는 이미 다른 페이지로 넘어간 뒤다. 실시간 빅데이터 분석의 요구사항은 결국 이 한 문장으로 요약된다 — 밀리초에서 수초 사이에 데이터를 수집·처리·분석해 즉시 의사결정을 지원해야 한다. 이걸 만족하려면 스트리밍 처리, 상태 관리, 정확히 한 번 처리(exactly once), 온라인 저장소 연계가 전부 갖춰져야 하고, 여기에 AI 모델 추론까지 같은 파이프라인 안에 넣으려는 시도가 Apache Spark를 중심으로 자리잡고 있다.

Spark가 스트리밍과 AI 워크로드를 한 번에 다루는 방식

Spark Structured Streaming은 마이크로배치 또는 Continuous Processing 중 하나를 선택해 적용할 수 있고, 대규모 데이터에 대한 분산 ETL·피처 엔지니어링·모델 추론을 하나의 실행 그래프로 통합할 수 있다. AI 워크로드는 Spark MLlib을 쓰거나, Pandas UDF·Arrow UDF로 ONNX·Torch·TF 같은 외부 모델을 직접 불러 추론하는 방식이 실무에서 흔하다. MLflow·Feast 같은 툴체인과 연계하고 GPU 리소스 스케줄링이나 RAPIDS 가속을 검토할 수도 있는데, 이 부분은 버전마다 지원 범위가 달라지므로 최신 정보를 확인해야 한다.

파이프라인을 이루는 구성 요소

데이터 파이프라인과 레이크하우스 기반이 출발점이다. Kafka·Pulsar·Kinesis 같은 이벤트 버스, CDC, 로그·텔레메트리를 소스로 받아 Delta Lake·Iceberg·Hudi나 온라인 캐시·키밸류 스토어로 흘려보낸다. 워터마크와 상태 저장소, 체크포인트로 지연 데이터를 처리하고 정확히 한 번 처리를 보장하며, 레이크하우스의 ACID 트랜잭션이 일관성을 지켜준다.

실시간 피처 엔지니어링·피처 스토어 단계에서는 스트리밍 집계, 윈도우, 상태 기반 조인으로 온라인 피처를 만들고 Feast나 Databricks Feature Store로 온라인·오프라인 피처의 일관성을 유지한다. 스키마 진화와 피처 버저닝, 유효기간 관리가 데이터 품질을 좌우한다.

모델 추론 전략(Streaming Inference)은 세 갈래로 나뉜다. 스트리밍 안에 내장한 추론(UDF·Pandas UDF), 외부 서빙(gRPC·HTTP) 호출, 오프라인 배치 스코어링과 캐싱을 섞는 방식이다. 지연 시간·처리량·운영 복잡도가 서로 트레이드오프 관계를 이루고, ONNX·TorchScript로 패키징하면 경량화와 이식성을 확보할 수 있다.

리소스·성능 최적화는 Adaptive Query Execution, Dynamic Allocation, Trigger 조절(ProcessingTime·AvailableNow)로 SLA를 맞추는 작업이다. 배치 크기, 동시성, 파티셔닝·코얼레스를 조정하고 GPU·CPU 혼합 스케줄링, 브로드캐스트·캐시 전략을 함께 적용한다.

관측성·신뢰성·데이터 품질은 Structured Streaming 지표와 지연·처리량 모니터링, 알림 설정으로 시작해 DLQ(Dead Letter Queue)로 불량 이벤트를 격리한다. Great Expectations나 Deequ로 데이터 품질을 검사하고 드리프트를 감지하며, 실험·모델 레지스트리와 연동해 재현성을 확보한다.

데이터가 흐르는 경로

이벤트가 Kafka 등으로 들어오면 스키마 검증·정규화를 거쳐 워터마크가 설정된다. 이후 상태 기반 집계와 피처 조회·조인을 거쳐 UDF나 외부 서빙으로 모델 추론이 실행되고, 품질 검사와 드리프트 모니터링이 뒤따른다. 결과는 Delta·키밸류·서빙 캐시에 정확히 한 번 기록되고 모니터링·알림으로 이어지며, 오류는 DLQ로 분리된다. 재시도 정책과 배치 격리, 누락 데이터 보정(재처리), 체크포인트·트랜잭션 일관성이 이 흐름 전체를 지탱한다.

⤢✕

추론 전략을 비교하면

패턴	성능(지연/처리량)	확장성	일관성	안정성	운영 편의
스트림 내장 추론(UDF/Pandas UDF)	지연이 짧고 처리량이 높음(실행기 스케일에 비례)	Spark 스케일아웃이 우수	체크포인트+트랜잭션으로 강한 일관성	코드/모델 배포 동기화가 필요	단일 스택이라 운영이 단순
외부 서빙 호출(gRPC/HTTP)	네트워크 오버헤드로 지연이 늘 수 있음	서빙/클러스터를 독립적으로 확장	네트워크/서빙 장애의 영향을 받음	서빙 이중화로 내결함성 향상	버전 롤백/카나리가 쉬움
오프라인 배치 스코어+캐시	온라인 지연이 최소지만 신선도 제약 존재	배치/캐시를 분리해서 확장	배치 기준의 결과 일관성	캐시 무효화가 복잡해질 수 있음	운영 구성이 단순

코드로 보면: Spark Structured Streaming + ONNX 추론

전제조건은 Spark 3.4+ (PySpark), Python 3.9~3.11, PyArrow 8+, onnxruntime 1.16+, delta-spark 3.0+다. Kafka를 소스로 쓰고 Delta Lake를 싱크로 삼아 체크포인트 기반 Exactly-once를 적용하며, 모델 파일은 ONNX 형식으로 드라이버·워커에서 공용 경로(HDFS·S3 등)에 접근 가능해야 한다.

# pyspark-shell or spark-submit with:
# --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.1,io.delta:delta-spark_2.12:3.0.0
# --conf spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, expr, pandas_udf
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
import numpy as np
import pandas as pd

spark = (SparkSession.builder
         .appName("rt-analytics-onnx")
         .config("spark.sql.streaming.schemaInference", "true")
         .config("spark.sql.adaptive.enabled", "true")
         .config("spark.sql.execution.arrow.pyspark.enabled", "true")
         .getOrCreate())

# 1) Kafka 소스
raw = (spark.readStream.format("kafka")
       .option("kafka.bootstrap.servers", "broker:9092")
       .option("subscribe", "events")
       .option("startingOffsets", "latest")
       .load())

# 2) 스키마 및 파싱
payload_schema = StructType([
    StructField("id", StringType(), False),
    StructField("f1", DoubleType(), True),
    StructField("f2", DoubleType(), True),
    StructField("f3", DoubleType(), True),
])

df = (raw.selectExpr("CAST(value AS STRING) as json_str")
      .select(from_json(col("json_str"), payload_schema).alias("j"))
      .select("j.*")
      .withWatermark("timestamp", "2 minutes") if "timestamp" in raw.columns else raw.select("json_str"))

# 3) ONNX 추론용 Pandas UDF
#    - 각 Executor 프로세스당 1회 로드(지연 초기화)
@pandas_udf("double")
def predict_udf(f1: pd.Series, f2: pd.Series, f3: pd.Series) -> pd.Series:
    import onnxruntime as ort
    import numpy as np
    global _sess
    if "_sess" not in globals():
        _sess = ort.InferenceSession("s3://bucket/models/model.onnx", providers=["CPUExecutionProvider"])
        global _input_name, _output_name
        _input_name = _sess.get_inputs()[0].name
        _output_name = _sess.get_outputs()[0].name
    X = np.stack([f1.to_numpy(float), f2.to_numpy(float), f3.to_numpy(float)], axis=1).astype(np.float32)
    y = _sess.run([_output_name], {_input_name: X})[0].reshape(-1)
    return pd.Series(y.astype(float))

scored = (df.withColumn("score", predict_udf(col("f1"), col("f2"), col("f3")))
            .withColumn("risk", expr("CASE WHEN score > 0.8 THEN 'high' WHEN score > 0.5 THEN 'mid' ELSE 'low' END")))

# 4) Delta Lake 싱크(Exactly-once)
query = (scored.writeStream
         .format("delta")
         .option("checkpointLocation", "s3://bucket/checkpoints/rt-analytics-onnx")
         .outputMode("append")
         .trigger(processingTime="5 seconds")
         .start("s3://bucket/delta/rt_scores"))

query.awaitTermination()

운영에서는 모델 경로를 버저닝(예: s3://…/model.onnx@vX)하고 롤링 업데이트 전략을 적용하며, 배포 전에는 Warm-up 배치를 돌려두는 것이 좋다. 외부 서빙을 쓴다면 foreachBatch 안에서 gRPC 배치 호출을 구현하고 실패·재시도·서킷브레이커를 붙여야 한다. 고정밀 Exactly-once가 필요하면 Kafka Transactional write나 Delta Lake의 멱등 키(upsert)를 쓴다.

실무에서는 이렇게 쓰인다

실시간 추천·랭킹에서는 클릭·뷰 스트림으로 피처를 갱신하고 유저·컨텍스트를 조인한 뒤 모델 추론을 수행해 수백 ms SLA를 맞춘다. 재훈련 주기를 단축하고 A/B 테스트 트래픽 셰이핑으로 실험을 가속하는 방식이 흔하다.

이상 탐지·부정 행위 탐지는 슬라이딩 윈도우 집계와 상태 기반 패턴 탐지를 결합해 임계를 초과하면 즉시 알림을 보내고, DLQ로 오탐 분석 루프를 구성한다. 모델 스코어와 규칙 기반 시그널을 앙상블하면 정밀도와 재현율의 균형을 잡을 수 있다.

제조·IoT 예지보전에서는 센서 텔레메트리를 실시간으로 집계해 피처를 만들고, 추론 결과를 경보·작업지시 시스템과 연계한다. 대역폭·보관 비용을 줄이기 위해 스키마 진화와 TTL 정책을 함께 적용한다.

실제로 얻는 것

환경에 따라 다르지만 p95 지연 시간이 30~~70% 개선되고 처리량이 2~~5배 확장되며 운영 인시던트 건수가 20~40% 감소한 사례가 보고된다. 정성적으로는 데이터-모델 일관성이 강화되고 실험·배포 주기가 짧아지며, 데이터 품질 가시성이 올라가면서 의사결정 신뢰성도 함께 높아진다.

보안·운영에서 챙겨야 할 트레이드오프

데이터·모델 보안은 전송 암호화(TLS), 비밀관리자(KMS·HashiCorp Vault), 모델 아티팩트 서명이 기본이지만 그만큼 오버헤드가 늘어난다. 신뢰성 확보를 위한 체크포인트·트랜잭션, 멱등 처리, 별도의 재처리 경로는 저장소 비용과 복잡성을 함께 끌어올린다. 리소스 최적화는 오토스케일·AQE·파티션 튜닝으로 비용 효율을 높이되, 극저지연이 필요한 구간은 외부 서빙과 경량 엔진 조합을 따로 검토하는 편이 낫다.

레이크하우스·상태 관리·피처 스토어·추론 전략을 유기적으로 결합하는 것이 Spark 기반 실시간 분석과 AI 워크로드 통합의 핵심이다. SLA에 맞는 추론 패턴을 고르고 일관성·관측성을 강화하면서 운영을 자동화해 TCO를 최적화하는 방향을 권장한다. 신규 기능이나 버전 호환성은 배포 전에 반드시 PoC로 검증해야 한다.