번역: HaUn Kim - Original Blog Link
리테일, 금융, 사이버 보안 및 기타 모든 산업에서는 비정상적인 행동이 발생하면 즉시 이를 감지하는 것이 절대적인 우선 순위입니다. 이를 감지할 수 있는 능력이 부족하면 매출 손실, 규제 기관으로부터의 벌금, 사이버 보안에서는 고객 개인정보 및 신뢰 침해 등의 문제가 발생할 수 있습니다. 그렇기 때문에 회사에서는 다소 비정상적인 신용카드 거래, 의심스러운 행동을 하는 사용자, 웹 서비스 요청에서 이상한 패턴을 식별하는 등의 이상을 탐지하는 것이 중요합니다. 이는 회사에서 좋은 하루를 보내는 것과 끔찍한 재앙을 맞이하는 것의 차이 만큼 엄청난 차이를 만들 수 있습니다.
이상 탐지에는 몇 가지 도전 과제가 있습니다. 첫 번째 도전은 '이상 징후 (anomaly)'가 무엇인지에 대한 데이터 사이언스적 질문입니다. 다행히도 머신 러닝은 데이터에서 정상 패턴과 비정상 패턴을 구별할 수 있는 강력한 도구입니다. 이상 탐지의 경우, 모든 이상 징후가 어떻게 생겼는지 알 수 없기 때문에 라벨이 지정되지 않은 데이터에서 비지도 학습을 사용하여 패턴을 학습해야 합니다.
비지도 머신러닝 모델을 사용하여 완벽한 이상 탐지 모델을 개발한다고 해도 이것은 겨우 시작에 불과합니다. 데이터가 원본 시스템에서 도착하는 즉시 각 관측이 수집되고 변환되어 모델에 점수를 매길 수 있도록 프로덕션에 모델을 적용하는 가장 좋은 방법은 무엇일까요? 또한 이를 실시간이나 5분에서 10분 정도의 짧은 간격으로 수행해야 한다면요? 이를 위해서는 정교한 추출, 로드, 변환(ELT) 파이프라인을 구축하고 비지도 머신러닝 모델을 통합하여 비정상 레코드를 정확하게 식별해야 합니다. 이러한 엔드투엔드 파이프라인은 데이터 품질을 보장하면서 항상 실행되는 프로덕션급이어야 하며, 기본 인프라를 유지 관리해야 합니다.
데이터브릭스를 사용하면 이러한 과정을 간단하게 처리할 수 있습니다. 데이터브릭스를 이용하면 실시간에 가까운 이상 탐지 파이프라인을 SQL로 구축하고, 머신러닝 모델 학습을 위해 Python을 사용할 수 있습니다. 데이터 수집, 변환, 모델 추론은 모두 SQL로 처리할 수 있습니다.
특히, 데이터브릭스 레이크하우스 플랫폼을 사용하면 비정상 레코드를 탐지하는 데 적합한 격리 포리스트 알고리즘(isolation forest algorithm)을 훈련하고, 훈련된 모델을 생성된 스트리밍 데이터 파이프라인에 통합하는 방법을 설명하는 블로그가 있습니다. 이 블로그에서는 데이터브릭스의 델타 라이브 테이블(DLT)을 사용하여 데이터 엔지니어링 프로세스를 자동화하고, 대규모 기본 인프라를 완벽하게 관리하는 방법을 소개하고 있습니다. 이를 통해 거의 실시간에 가까운 이상 탐지 시스템을 구축할 수 있습니다. 이 블로그에서 사용된 데이터는 신용카드 거래를 시뮬레이션하기 위해 생성된 합성 데이터의 샘플이며, 이렇게 탐지된 이상 거래는 사기 거래입니다.
블로그에 소개된 ML 및 델타 라이브 테이블 기반 이상 탐지 솔루션의 아키텍처 개요
Scikit-learn 격리 포리스트 알고리즘 구현은 기본적으로 데이터브릭스 머신러닝 런타임에서 사용할 수 있으며, MLflow 프레임워크를 사용하여 학습되는 동안 이상 탐지 모델을 추적하고 로깅합니다. ETL 파이프라인은 전적으로 델타 라이브 테이블을 사용하여 SQL로 개발됩니다.
격리 포리스트는 랜덤 포리스트와 유사한 트리 기반 앙상블 알고리즘의 한 유형입니다. 이 알고리즘은 주어진 관측값 집합의 아웃라이어보다 인라이어를 분리하기가 더 어렵다고 가정하도록 설계되었습니다. 즉, 일반적인 지점들(예: 일반적인 신용카드 거래)은 분리하기 어려워 의사 결정 트리에서 더 깊은 위치에 있을 것이고, 비정상적인 지점들은 그 반대의 경우입니다. 이 알고리즘은 레이블이 없는 관찰 집합에 대해 학습한 후, 이전에 볼 수 없었던 데이터에서 비정상적인 기록을 예측하는 데 사용할 수 있습니다.
데이터브릭스에서 머신러닝과 관련된 작업을 할 때는 머신러닝 런타임과 함께 클러스터를 사용하는 것이 필수입니다. 데이터 사이언스 및 머신러닝 관련 작업에 일반적으로 사용되는 많은 오픈 소스 라이브러리는 ML 런타임에서 기본적으로 사용할 수 있습니다. Scikit-learn은 이러한 라이브러리 중 하나이며, 격리 포리스트 알고리즘을 훌륭하게 구현한 라이브러리입니다.
모델을 정의하는 방법은 다음과 같습니다.
from sklearn.ensemble import IsolationForest
isolation_forest = IsolationForest(n_jobs=-1, warm_start=True, random_state=42)이 런타임은 주로 머신러닝 실험 추적, 모델 스테이징, 그리고 배포를 위해 노트북 환경과 MLflow를 긴밀하게 통합할 수 있도록 지원합니다.
ML 클러스터에 연결된 노트북 환경에서 수행되는 모든 모델 훈련 또는 하이퍼파라미터 최적화는 기본적으로 활성화된 MLflow 자동 로깅 기능을 통해 자동으로 기록됩니다.
모델이 로깅되면 MLflow 내에서 다양한 방법으로 모델을 등록하고 배포할 수 있습니다. 특히, 아래 이미지에서 볼 수 있는 것처럼 MLflow는 사용자 인터페이스(UI) 내에서 벡터화된 사용자 정의 함수(User Defined Function, UDF)로 배포하기 위한 코드를 생성하여 Apache Spark™를 통한 분산 인스트림 또는 배치 추론을 지원합니다. 이를 통해 MLflow는 모델을 손쉽게 UDF로 생성하고 등록할 수 있도록 도와줍니다.
MLflow는 모델 추론을 위해 Apache Spark UDF를 생성하고 등록하기 위한 코드를 생성합니다
더불어, MLflow REST API를 사용하면 몇 줄의 코드로 기존 운영 중인 모델을 아카이브하고 새로 학습된 모델을 운영 환경에 배포할 수 있습니다.
def train_model(mlFlowClient, loaded_model, model_name, run_name)->str:
"""
Trains, logs, registers and promotes the model to production. Returns the URI of the model in prod
"""
with mlflow.start_run(run_name=run_name) as run:
# 0. Fit the model
loaded_model.fit(X_train)
# 1. Get predictions
y_train_predict = loaded_model.predict(X_train)
# 2. Create model signature
signature = infer_signature(X_train, y_train_predict)
runID = run.info.run_id
# 3. Log the model alongside the model signature
mlflow.sklearn.log_model(loaded_model, model_name, signature=signature, registered_model_name= model_name)
# 4. Get the latest version of the model
model_version = mlFlowClient.get_latest_versions(model_name,stages=['None'])[0].version
# 5. Transition the latest version of the model to production and archive the existing versions
client.transition_model_version_stage(name= model_name, version = model_version, stage='Production', archive_existing_versions= True)
return mlFlowClient.get_latest_versions(model_name, stages=["Production"])[0].source프로덕션 시나리오에서는 단일 레코드가 모델에 한 번만 스코어되는 것이 중요합니다. 데이터브릭스에서는 자동 로더를 사용하여 이 "정확히 한 번" 동작을 보장할 수 있습니다. 자동 로더는 Python 또는 SQL을 사용하는 델타 라이브 테이블과 구조화된 스트리밍 애플리케이션과 함께 작동합니다.
또 다른 중요한 고려 요소는 비정상적인 이벤트의 특성이 시간에 따라 변한다는 것입니다. 따라서 새로운 데이터가 도착할 때마다 모델을 재학습해야 합니다.
데이터브릭스 워크플로우에서 모델 학습 로직이 포함된 노트북을 예약된 작업으로 생성하면, 작업이 실행될 때마다 최신 모델을 효과적으로 재학습하고 프로덕션에 투입할 수 있습니다.
머신러닝 측면에서는 이러한 과제가 일부에 불과합니다. 데이터 수집, 변환, 모델 추론을 결합한 실시간에 가까운 프로덕션 수준의 데이터 파이프라인을 구축하는 것이 더 어려운 과제입니다. 이 프로세스는 복잡하고 시간이 많이 소요되며 오류가 발생하기 쉽습니다.
이를 상시 가동할 수 있는 인프라를 구축하고 유지 관리하며 오류를 처리하려면 데이터 엔지니어링보다 더 많은 소프트웨어 엔지니어링 노하우가 필요합니다. 또한 전체 파이프라인을 통해 데이터 품질을 보장해야 합니다. 특정 애플리케이션에 따라 복잡성이 추가될 수 있습니다.
여기서 델타 라이브 테이블(DLT)이 등장합니다.
DLT 용어로 노트북 라이브러리는 기본적으로 DLT 파이프라인의 코드 일부 또는 전체가 포함된 노트북입니다. DLT 파이프라인에는 하나 이상의 노트북이 연결될 수 있으며, 각 노트북은 SQL 또는 Python 구문을 사용할 수 있습니다. 첫 번째 노트북 라이브러리에는 파이썬으로 구현된 로직이 포함되어 있으며, MLflow 모델 레지스트리에서 모델을 가져오고, 수집된 레코드가 파이프라인의 다운스트림에서 기능화되면 모델 추론 함수를 사용할 수 있도록 UDF를 등록합니다. 유용한 팁: DLT Python 노트북에서는 새 패키지를 설치하기 위해 첫 번째 셀에 %pip magic 명령을 사용해야 합니다.
두 번째 DLT 라이브러리 노트북은 Python 또는 SQL 구문으로 구성할 수 있습니다. DLT의 다재다능함을 증명하기 위해 SQL을 사용하여 데이터 수집, 변환 및 모델 추론을 수행했습니다. 이 노트북에는 파이프라인을 구성하는 실제 데이터 변환 로직이 포함되어 있습니다.
수집은 객체 스토리지로 스트리밍된 데이터를 점진적으로 로드할 수 있는 자동 로더로 수행됩니다. 이는 메달리온 아키텍처의 브론즈(raw data) 테이블로 읽혀집니다. 또한 아래 구문에서 스트리밍 라이브 테이블은 오브젝트 스토리지에서 데이터가 지속적으로 수집되는 곳입니다. 자동 로더는 데이터가 수집될 때 스키마를 감지하도록 구성됩니다. 자동 로더는 진화하는 스키마도 처리할 수 있으며, 이는 많은 실제 이상 탐지 시나리오에 적용됩니다.
CREATE OR REFRESH STREAMING LIVE TABLE transaction_readings_raw
COMMENT "The raw transaction readings, ingested from landing directory"
TBLPROPERTIES ("quality" = "bronze")
AS SELECT * FROM cloud_files("/FileStore/tables/transaction_landing_dir", "json", map("cloudFiles.inferColumnTypes", "true"))또한 DLT를 사용하면 데이터 품질 제약 조건을 정의할 수 있으며, 개발자나 분석가가 오류를 수정할 수 있는 기능을 제공합니다. 만약 주어진 레코드가 주어진 제약 조건을 충족하지 않는다면, DLT는 해당 레코드를 유지 또는 삭제하거나 파이프라인을 완전히 중단할 수 있습니다. 아래의 예시에서는 트랜잭션 시간이나 금액이 지정되지 않은 경우, 레코드를 삭제하는 변환 단계 중 하나에 제약 조건이 정의되어 있습니다.
CREATE OR REFRESH STREAMING LIVE TABLE transaction_readings_cleaned(
CONSTRAINT valid_transaction_reading EXPECT (AMOUNT IS NOT NULL AND TIME IS NOT NULL) ON VIOLATION DROP ROW
)
TBLPROPERTIES ("quality" = "silver")
COMMENT "Drop all rows with nulls for Time and store these records in a silver delta table"
AS SELECT * FROM STREAM(live.transaction_readings_raw)델타 라이브 테이블은 사용자 정의 함수(UDF)도 지원합니다. UDF는 SQL을 사용하여 스트리밍 DLT 파이프라인에서 모델 추론을 활성화하는 데 사용할 수 있습니다. 아래 예시에서는 훈련된 격리 포레스트 모델을 캡슐화하는 이전에 등록된 Apache Spark™ 벡터화된 UDF를 사용하고 있습니다.
CREATE OR REFRESH STREAMING LIVE TABLE predictions
COMMENT "Use the isolation forest vectorized udf registered in the previous step to predict anomalous transaction readings"
TBLPROPERTIES ("quality" = "gold")
AS SELECT cust_id, detect_anomaly(<enter by="" column="" commas="" names="" separated="">) as
anomalous from STREAM(live.transaction_readings_cleaned)데이터 사이언티스트가 Python으로 훈련한 머신 러닝 모델(예: scikit-learn, xgboost 또는 다른 머신 러닝 라이브러리)을 사용하여 전체 SQL 데이터 파이프라인에서 추론할 수 있으므로, SQL 분석가 및 데이터 엔지니어에게는 매우 흥미로운 기능입니다.
이 노트북은 DLT 파이 프라인을 생성하는 데 사용됩니다. 구성 세부 정보는 아래의 섹션에서 자세히 설명되어 있습니다. 리소스를 설정하고 테이블을 구성하며 종속성을 파악하는 짧은 시간이 지나면, 학습된 머신 러닝 모델을 통해 데이터가 지속적으로 처리되고 거의 실시간으로 비정상적인 레코드가 감지되는 DLT 파이프라인이 UI에 렌더링됩니다. 이는 DLT가 추상화하는 다른 복잡한 작업을 최종 사용자로부터 감추어줍니다.
DLT 사용자 인터페이스에서 본 엔드 투 엔드 델타 라이브 테이블 파이프라인
이 파이프라인이 실행되는 동안, 데이터브릭스 SQL을 사용하여 식별된 비정상 레코드를 시각화할 수 있으며, 데이터브릭스 SQL 대시보드의 새로고침 기능을 통해 지속적으로 업데이트할 수 있습니다. '예측' 테이블에 대해 실행된 쿼리를 기반으로 시각화된 대시보드를 아래에서 확인할 수 있습니다.
예측된 비정상 레코드를 대화형으로 표시하도록 구축된 Databricks SQL 대시보드
이 블로그에서는 데이터브릭스 머신러닝과 워크플로우를 사용하여 이상 탐지를 위한 격리 포리스트 알고리즘을 훈련하는 데 사용할 수 있는 기능과, 거의 실시간으로 이러한 기능을 수행할 수 있는 델타 라이브 테이블 파이프라인을 정의하는 프로세스에 대해 자세히 설명합니다. 델타 라이브 테이블은 최종 사용자로부터 프로세스의 복잡성을 추상화하여 자동화하는 기능을 제공합니다.
이 블로그에서는 델타 라이브 테이블의 일부 기능만을 소개하였지만, 데이터브릭스의 주요 기능에 대한 이해하기 쉬운 설명서는 https://docs.databricks.com/data-engineering/delta-live-tables/index.html 에서 확인하실 수 있습니다. 해당 문서에서 델타 라이브 테이블의 전체 기능을 자세히 알아볼 수 있습니다.
데이터브릭스 워크플로우 사용자 인터페이스를 사용하여 델타 라이브 테이블 파이프라인을 생성할 수 있습니다
실시간에 가까운 방식으로 이상 탐지를 수행하기 위해서는 DLT(델타 라이브 테이블) 파이프라인을 연속 모드에서 실행해야 합니다. 이 블로그의 리포지토리에서는 앞서 설명한 Python 및 SQL 노트북을 사용하여, 공식 빠른 시작 페이지에 설명된 프로세스에 따라 생성할 수 있습니다. 물론 다른 구성도 원하는 대로 입력할 수 있습니다.
데이터의 이상 탐지를 위해 일괄 수집한 레코드와 같은 경우, 간헐적인 파이프라인 실행도 가능합니다. 예를 들어, 파이프라인을 트리거 모드에서 10분 정도의 짧은 간격으로 실행할 수 있습니다. 그리고 이 트리거된 파이프라인이 실행될 일정을 지정하고, 각 실행에서 데이터는 파이프라인을 통해 증분 방식으로 처리됩니다.
이후, 파이프라인을 시작하고 데이터를 처리하는 동안 다양한 레코드 부하를 처리하기 위해 클러스터 자동 확장이 활성화된 파이프라인 구성을 저장할 수 있습니다. 또는 이러한 모든 구성을 JSON 형식으로 깔끔하게 설명하여 동일한 입력 양식에 입력할 수도 있습니다.
델타 라이브 테이블은 클러스터 구성, 기본 테이블 최적화, 그리고 최종 사용자를 위한 기타 여러 가지 중요한 세부 정보를 고려합니다. 파이프라인을 실행하기 위해 개발 모드(반복 개발에 도움) 또는 프로덕션 모드(프로덕션 환경에 맞춤)를 선택할 수 있습니다. 후자의 경우, DLT는 자동으로 재시도 및 클러스터 재시작을 수행합니다.
마지막으로, 이 블로그에서 언급한 모든 작업은 델타 라이브 테이블 REST API를 통해 수행할 수 있습니다. 이는 격리 포리스트가 재훈련될 때마다 다운타임 없이 연속 모드에서 실행되는 DLT 파이프라인을 즉시 편집할 수 있는 프로덕션 시나리오에 특히 유용합니다.
이 예제에서 델타 라이브 테이블 파이프라인에 대한 구성. 생성된 델타 테이블을 저장할 대상 데이터베이스 이름을 입력합니다
이 솔루션을 다시 만들기 위한 노트북과 단계별 지침은 모두 리포지토리에 포함되어 있습니다:
이 노트북과 지침을 사용하여 모델 학습 작업을 수행하기 위해서는 데이터브릭스 머신러닝 런타임이 포함된 클러스터를 사용해야 합니다. 여기에 제공된 예는 상대적으로 단순하지만, 더 복잡한 변환에도 동일한 원칙이 적용될 수 있으며, 델타 라이브 테이블은 이러한 파이프라인 구축의 복잡성을 줄이기 위해 설계되었습니다. 여러분의 사용 사례에 맞게 이 블로그의 아이디어를 조정해 보시기 바랍니다.
추가로, DLT 기능에 대한 훌륭한 데모와 워크스루는 영상에서 확인할 수 있습니다:
