AWS 기술 블로그

티머니의 MLOps 구현 사례 : Amazon SageMaker를 활용한 배차모델 자동화 및 배포

이 블로그는 티머니의 반용주 매니저, 구현서 매니저, 오지훈 매니저와 함께 작성되었습니다

티머니 로고 티머니는 ‘이동을 편하게, 세상을 이롭게’를 경영철학으로, ‘더 편한 이동과 결제를 위한 플랫폼 기업’으로 성장하고 있습니다.

티머니는 세계적으로 기술력을 인정 받고 있는 ‘티머니 교통카드 시스템’을 기반으로 대중교통 정산 및 모바일티머니 서비스를 제공하며, 뉴질랜드, 말레이시아, 몽골 등 전 세계에 교통카드 시스템을 수출하고 있습니다. 티머니는 ‘고객’과 ‘공익성’을 최우선 가치로 모빌리티와 페이먼트 영역에서 산업을 선도하고 있습니다. 모빌리티 영역에서는 ‘티머니GO’를 통해 대중교통 중심의 통합 이동 서비스로 ‘더 편한 이동’을 선도하고 있으며, 페이먼트 영역에서는 ‘모바일티머니’, ‘기후동행카드’ 및 ‘태그리스 결제’와 같은 혁신 서비스로 시민들의 ‘더 편한 결제’를 위해 노력하고 있습니다.

특히, 누적회원 1,100만명 모빌리티 슈퍼앱 ‘티머니GO’ 에서 제공하는 택시 호출 서비스 티머니GO온다택시는 AI최적 배차/승차거부 없는 탑승 등으로 택시 탑승 문화를 선진화하며, 창원, 춘천, 원주, 대전, 양주, 제주, 서산 등 지역 브랜드 콜택시와의 상생은 물론 교통약자, 디지털 소외계층을 위한 다양한 서비스를 통해 ‘지역상생 착한택시’로 자리잡고 있습니다.

이번 게시물에서는 티머니GO온다택시의 호출 성공율을 높이기 위해 ML모델을 개발하고, 예측 정확도 유지를 위해 티머니에서 진행한 다양한 노력과 프로세스 혁신의 과정을 소개합니다. 특히, 온프레미스에서 자체 개발한 머신러닝(ML) 모델을 Bring Your Own Model(BYOM) 방식으로 배포하여 Amazon SageMaker 기반에서 서비스한 부분과, MLOps를 활용하여 머신러닝 라이프사이클을 자동화한 여정을 다룹니다.

기존 서비스의 한계와 새로운 접근 방식

기존 택시 호출 서비스는 출근, 심야시간과 같이 특정 시간에 수요가 몰리는 특징이 있었습니다. 티머니 온다택시는 이와 같이 급격히 늘어나는 수요에 대응하고, 기사님들의 호출 수락을 독려하기 위해 해당시간의 모든 호출에 대해 기사 마일리지를 제공했습니다.

그림 1. 온다택시 기사 마일리지 제공 서비스 플로우

이 방식은 단기적으로 기사의 호출 수락율을 상승시키는 효과가 있었지만, 한정적인 예산안에서 모든 호출에 대해 마일리지를 제공한다는 것은 불가능했습니다. 이에 성공할 확률이 낮은 호출을 판단해 마일리지를 지급하는 방안이 제시되었고, 시간대, 위치, 교통상황, 날씨 등을 고려해 호출 별 실시간 성공 확률을 예측하는 ML모델 개발을 계획하게 되었습니다.

초기 ML모델은 온프레미스의 로컬 환경에서 개발했습니다. 로컬에 저장된 모델로 평가용 데이터의 전처리 후, 요청 할 때 마다 로컬의 파이썬 코드를 실행하여 예측 결과를 얻는 방식이었습니다.

하지만, 프로젝트의 목적 상 호출에 대한 실시간 예측과 주기적인 데이터 전처리, 모델 학습 및 배포가 필요했는데, 로컬의 모델을 활용하는 방식은 프로젝트의 목적과 부합하지 않았습니다.

더불어 자동화를 위해서는 MLOps가 필요하다고 판단했습니다. 하지만 사내에 MLOps 경험자가 없고, 이를 공부하며 진행하기에는 인원이 부족한 상황이었습니다.

ML모델 재학습 및 재배포를 위한 Amazon SageMaker 선택

티머니는 이미 로컬 환경에서 개발하고 학습한 예측 모델을 보유하고 있었습니다. 따라서 이번 프로젝트의 목표는 이 모델의 성능을 유지하면서 클라우드 환경으로 마이그레이션하고, 자동화된 CI/CD 파이프라인을 구축하는 것이었습니다. 아울러 아래와 같은 사항을 고려했습니다.

  • Amazon SageMaker의 완전 관리형 서비스 특성은 MLOps 전문 인력이 부족한 상황에서도 큰 장점을 제공했습니다. 모델 구축부터 학습, 배포에 이르는 전 과정을 효율적으로 관리할 수 있어 SageMaker 도입을 결정하게 되었습니다.
  • 티머니는 SageMaker Script Mode는 BYOM(Bring Your Own Model) 접근 방식을 채택했습니다. Script Mode를 활용하면 AWS에서 제공하는 베이스 모델 컨테이너(XGBoost, PyTorch, TensorFlow 등)를 기반으로, 기존 모델을 활용하면서도 티머니의 자체 학습 및 추론 코드를 그대로 사용할 수 있었습니다. 이 방식은 기존에 개발한 ML모델을 클라우드 환경으로 신속하게 마이그레이션할 수 있어 모델 배포 관리에 대한 부담을 최소화하면서도 지속적인 모델 개선과 효율적인 운영을 가능하게 했습니다.
  • SageMaker Notebook Jobs를 활용하여, 모델 투입 변수 업데이트 및 학습 작업을 자동화할 수 있었습니다. 온다택시의 마이그레이션 된 ML모델은 매월 지속적으로 업데이트가 필요합니다. 특히 수요와 공급을 대표하는 데이터의 분포 변화는 택시 호출의 성공확률에 큰 영향을 미치기 때문에, 이를 놓치지 않고 모델에 반영하는 것이 중요했습니다.
  • SageMaker Pipelines을 구축하여 주기적인 모델 학습, 배포, 업데이트 및 버전 관리를 가능하게 했습니다.
  • SageMaker Endpoint를 생성해서 배포된 모델에 대해 실시간으로 추론할 수 있었습니다. 프로젝트의 목적이 택시 호출에 대해 실시간으로 평가를 하고 조치를 취하는 것이기 때문에 SageMaker Endpoint가 중요한 역할을 했습니다.

솔루션 아키텍쳐와 흐름도

티머니는 로컬에서 모델 개발 당시 가장 좋은 성능을 보인 XGBoost 모델을 사용했습니다. XGBoost는 지도학습 기반 그래디언트 부스팅 알고리즘을 사용하는 대중적인 오픈소스 ML모델입니다.

XGBoost 알고리즘을 사용하기 위해서는 Amazon SageMaker AI의 내장 알고리즘 또는 로컬 환경에서 학습 스크립트를 실행하기 위한 프레임워크로 사용할 수 있습니다.

BYOM을 진행할 때 Amazon SageMaker XGBoost Bring Your Own Model 방식을 활용하여 학습된 모델을 S3로 업로드하고 SageMaker Endpoint에서 호스팅하는 방법도 가능합니다. 티머니는 커스텀 학습 및 추론 코드를 AWS 클라우드에 올려서 모델을 매월 새로 학습, 배포하는 형태로 진행했습니다.

이후 MLOps 파이프라인을 구성하여 주기적인 데이터 전처리, 모델 학습, 평가, 모델 등록, 그리고 엔드포인트 배포까지의 모든 과정을 자동화했습니다. 이 흐름은 Amazon SageMaker Pipelines, Amazon SageMaker Processing Job, SageMaker Training Job, 그리고 SageMaker Endpoint를 이용해 단계별로 이루어졌으며, 모든 단계가 유기적으로 연결되어 전체적인 ML 라이프사이클을 효율적으로 관리할 수 있도록 설계되었습니다.

그림2. 티머니 온다택시 MLOps 솔루션 아키텍처

아래는 SageMaker Studio의 Visual Designer 기능을 활용하여 ML 워크플로우를 시각화한 화면 입니다.

그림3. SageMaker에서 생성한 ML 워크플로우

위와 같이 파이프라인이 배포된 후에 SageMaker Studio를 활용하여 파이프라인을 Directed Acyclic Graph(DAG) 형태로 시각화할 수 있고 관리할 수 있습니다.

SageMaker BYOM 및 Pipeline 구현 상세

다음 섹션에서는 SageMaker Pipeline을 활용한 Pipeline 개발과정을 샘플 코드와 함께 단계 별로 소개합니다.

1. 로컬 모델 개발 평가

최초 모델 개발은 로컬 환경에서 시작했습니다. 이 과정에서 주력한 부분은 다음과 같습니다:

  • 변수 생성 및 테스트: 호출 성공확률에 영향을 미치는 수요와 공급 관련 변수들을 생성하고 테스트했습니다.
  • 모델 개발 및 평가: 유의미한 변수들을 선별하여 모델에 투입하고, 개발 및 평가 과정을 거쳤습니다.
  • 성능 최적화: 특히 두 가지 핵심 지표에 주목했습니다.
    • Recall Score: 실제 성공한 호출을 정확히 예측하는 능력을 평가합니다.
    • Precision Score: 불필요한 비용 지출을 줄이기 위한 정확도를 평가합니다.

티머니는 이 두 지표를 만족스러운 수준으로 높이기 위해 여러 차례의 반복 작업을 수행했습니다. 최종적으로, 개발한 모델은 Hyper-Parameter Optimization (HPO) 과정 없이도 충분히 만족스러운 성능을 보여주었습니다. 이로써 로컬환경의 모델 개발 과정은 마무리 지었습니다.

로컬환경에서 모델 개발을 위해 작성했던 데이터 전처리, 학습/평가 데이터 생성, 모델 개발 파이썬 코드는 SageMaker Pipeline 생성을 위해 S3 버킷의 작업경로에 저장했습니다.

2. SageMaker Pipeline 생성

SageMaker Pipeline은 머신러닝 워크플로우를 자동화하고 관리하는 강력한 도구입니다. SageMaker Pipeline은 UI, SageMaker Python SDK 또는 SageMaker Pipeline Definition JSON 스키마를 사용하여 빌드할 수 있습니다.

초기에 티머니는 AWS 환경에서 ML모델 학습 및 배포를 SageMaker Pipeline 없이 진행했습니다. 그러나 머신러닝 모델을 매월 주기적으로 재학습하고 배포해야하는 요건이 있었기 때문에 MLOps를 초기 ML모델 배포 성공 이후 검토하여 적용했습니다.

티머니는 SageMaker Pipeline 생성을 처음 시도했을 때, 여러 가지 문제에 직면했습니다. 특히 스크립트 파일 관리와 각 파이프라인 단계에서의 스크립트 사용에 어려움을 겪었습니다. 여러 시도 끝에 Notebook Job을 활용한 방법으로 성공할 수 있었습니다.

주요 단계는 다음과 같습니다:

  • 스크립트 준비: ‘xgboost_starter_script.py’ 파일을 메인 entry_point로 사용했습니다.
  • 스크립트 압축: 이 파일을 .tar.gz 형식으로 압축했습니다.
  • S3 업로드: 압축된 파일을 S3 버킷에 업로드했습니다.
  • Pipeline 설정: source_dir 파라미터에 ‘S3버킷주소/압축파일명.tar.gz’ 형식으로 경로를 지정했습니다.

이 방법을 train, evaluation, deploy 단계에 모두 일관되게 적용했더니 Pipeline 생성에 성공했습니다.

3. 변수(Feature) 생성 단계

SageMaker Processing Job은 SageMaker 의 완전 관리형 인프라에서 데이터 사전 및 사후 처리, 기능 엔지니어링 및 모델 평가 작업을 실행할 수 있습니다.

티머니는 Processing Job을 활용하여 AWS 클라우드에 저장된 Raw data를 활용하여 모델에 투입할 변수를 생성하였습니다. 호출 성공 확률을 예측하기 위한 여러 변수들을 생성하기 위해, Scikit-Learn 기반의 데이터 전처리 스크립트를 사용했습니다. 이 과정에서 처리해야 하는 데이터의 크기에 따라 인스턴스 타입을 유연하게 변경하여 효율적으로 처리했습니다.

또한, 데이터 전처리를 위한 변수 생성 단계에서는 빅데이터팀 이외에도 엔드포인트를 실제로 사용하는 애플리케이션팀과의 협업이 중요했으며, 데이터의 버전 관리를 통해 추적 가능성을 높였습니다. 생성된 변수들은 Amazon Simple Storage Service (Amazon S3)에 저장되어 이후 학습 및 평가 데이터 생성 단계에서 사용되었습니다.

아래는 SageMaker Processing Job을 활용한 변수 생성 샘플 코드입니다. 이 코드는 주어진 파일을 사용하여 변수 생성 작업을 수행하는 SageMaker Pipeline의 단계를 생성합니다. 이 단계는 지정된 스크립트를 실행하고, 결과를 S3에 저장하며, 필요한 설정(인스턴스 유형, 역할 등)을 포함합니다. 이는 주로 데이터 전처리나 특성 엔지니어링과 같은 작업에서 활용됩니다.

티머니는 변수가 여러 개이기 때문에 변수 생성 python 파일을 file이라는 매개변수를 입력하여 각 변수를 생성했습니다. 기본적으로 인스턴스는 ‘ml.m5.2xlarge’를 사용하는데 이 과정에서는 처리하는 데이터의 크기가 커서 ‘ml.m5.4xlarge’로 유연하게 적용하였습니다.

def _step_variable_generation(self, file):
        strInstanceType = "ml.m5.4xlarge"
        nInstanceCount = 1
        strProcPrefixPath = "/opt/ml/processing"        

        strOutputPath = os.path.join(
            "s3://{}".format(self.strBucketName),
            'dataset',
            yyyymm,
            "feature"
        )      

        script_name = file.split('.')[1].strip()
        
        processing_processor = SKLearnProcessor(
            framework_version='0.23-1',
            role=self.strExecutionRole,
            instance_count=nInstanceCount,
            instance_type=strInstanceType,
            base_job_name="variable-generation",
            sagemaker_session=self.pipeline_session,
        )

        step_processing_args = processing_processor.run(
            code=f'{self.strJobInput}/source_python_file/{file}',
            inputs=[
                ProcessingInput(
                    source=f'{self.strJobInput}/source_python_file/requirements/requirements.txt',
                    input_name="requirements-txt",
                    destination=os.path.join(strProcPrefixPath, "input")
                )
            ],
            outputs=[
                ProcessingOutput(
                    source=os.path.join(strProcPrefixPath, "output"),
                    destination=strOutputPath,
                )
            ],
            job_name=f"variable-generation-{script_name}"
)
processing_step = ProcessingStep(
            name=f"VariableGeneration_{script_name}",
            step_args=step_processing_args,
            cache_config=self.cache_config,
            retry_policies=self.retry_policies
        )

        return processing_step
        )
Python

4. 데이터 전처리 단계 : 학습 테스트 데이터 세트 생성

다음은 이전 변수 생성 단계를 통해 S3에 저장된 각 변수 데이터를 기반으로, 학습 및 테스트 데이터세트를 생성하는 단계입니다. 이 단계는 특성(Feature) 생성 단계(variable_generation_feature_1, variable_generation_feature_2 등)에 의존하고 있어, 이전 단계가 완료된 후에 실행됩니다. ProcessingStepdepends_on에 변수 생성 단계들을 포함하여, 모든 변수 생성이 완료된 후 data 생성을 시작합니다.

입력으로 requirements.txt 파일을 사용하여 필요한 라이브러리를 설치합니다. 출력으로 학습 데이터와 테스트 데이터를 별도의 S3 경로에 저장합니다. 이 이후부터는 인스턴스가 이전 단계보다 리소스가 덜 필요로 하기 때문에 ‘ml.m5.2xlarge’로 변경합니다.

def _step_data_processing(self):
        strInstanceType = "ml.m5.2xlarge"
        nInstanceCount = 1
        strProcPrefixPath = "/opt/ml/processing"        

        strOutputPath = os.path.join(
            "s3://{}".format(self.strBucketName),
            'dataset'
        )      

        processing_processor = SKLearnProcessor(
            framework_version='0.23-1',
            role=self.strExecutionRole,
            instance_count=nInstanceCount,
            instance_type=strInstanceType,
            base_job_name="creating-train-test-data",
            sagemaker_session=self.pipeline_session,
        )

        step_processing_args = processing_processor.run(
            code=f'{self.strJobInput}/source_python_file/creation_train_test_data.py',
            inputs=[
                ProcessingInput(
                    source=f'{self.strJobInput}/source_python_file/requirements/requirements.txt',
                    input_name="requirements-txt",
                    destination=os.path.join(strProcPrefixPath, "input")
                )
            ],
            outputs=[
                ProcessingOutput(
                    output_name='train',
                    destination=f'{strOutputPath}/train',
                    source='/opt/ml/processing/output/train'
                ),
                ProcessingOutput(
                    output_name='test',
                    destination=f'{strOutputPath}/test',
                    source='/opt/ml/processing/output/test'
                )                
            ],
            job_name="data-processing",
        )

        self.data_processing_step = ProcessingStep(
            name="DataProcessing",
            step_args=step_processing_args,
            cache_config=self.cache_config,
            retry_policies=self.retry_policies,
            depends_on=[self.variable_generation_feature_1, self.variable_generation_feature_2]
        )
Python

5. 모델 학습, 평가 저장 단계

아래는 SageMaker Pipeline을 활용한 모델 학습, 평가 및 저장 과정을 설명합니다. (일부 정보는 고객사 주요 정보로서 생략하였습니다.)

모델 학습 스크립트

다음은 학습 단계로서 이전에 생성된 학습/평가 데이터 세트를 활용합니다. ML모델 학습은 XGBoost Estimator를 사용하여 사용자 지정 학습 스크립트를 구성했습니다. 하이퍼파라미터는 이전 로컬 개발의 평가 결과를 바탕으로 고정된 값을 사용하여 XGBoost 모델에 딕셔너리 형태로 정의합니다.

SageMaker Training Job을 사용하면 학습 스크립트에서 사용자 지정 지표를 게시하여 시간 경과에 따른 교육 진행 상황을 추적할 수 있습니다.이러한 지표는 SageMaker Experiment, SageMaker Console 및 Amazon CloudWatch를 통해 직접 볼 수 있습니다.

def _step_training(self, ):

        strInstanceType = "ml.m5.2xlarge"
        nInstanceCount = 1
        bSpotTraining = False
        
        train_path = os.path.join(self.strDataPath, 'train')
        test_path = os.path.join(self.strDataPath, 'test')
        s3_input_train = sagemaker.inputs.TrainingInput(s3_data=f'{train_path}/train.csv')
        s3_input_test = sagemaker.inputs.TrainingInput(s3_data=f'{test_path}/test.csv')    

        inputs = {'training': s3_input_train, 'testing': s3_input_test}
            
        strOutputPath = os.path.join(
            "s3://{}".format(self.strBucketName),
            self.strPipelineName,
            "training",
            "model-output"
        )

        strCodeLocation = os.path.join(
            "s3://{}".format(self.strBucketName),
            self.strPipelineName,
            "training",
            "backup_codes"
        )
       
        self.estimator = XGBoost(
            entry_point='xgboost_starter_script.py',
            source_dir=f'{self.strSource}/source/train/sourcedir.tar.gz',
            output_path=strOutputPath,
            code_location=strCodeLocation,
            hyperparameters=dicHyperparameters,
            role=self.strExecutionRole,
            instance_count=nInstanceCount,
            instance_type=strInstanceType,
            framework_version="1.7-1",
            max_run=nMaxRun,
            use_spot_instances=bSpotTraining,
            max_wait=nMaxWait,
            enable_sagemaker_metrics=True,            
            sagemaker_session=self.pipeline_session
            
        )
        
        job_name = "-".join([self.strPipelineName, "training-job"])
        step_training_args = self.estimator.fit(
            inputs=inputs,
            job_name=job_name,
            experiment_config={
              'TrialName': job_name,
              'TrialComponentDisplayName': job_name,
            },
            logs="All",
        )
        
        self.training_process = TrainingStep(
            name="TrainingProcess",
            step_args=step_training_args,
            depends_on=[self.data_processing_step],
            cache_config=self.cache_config,
            retry_policies=self.retry_policies
        )
Python

모델 평가 스크립트

아래 샘플 코드는 Processing Step을 사용하여 모델을 평가하는 방법을 보여줍니다.

Processing Step에서는 학습 단계의 프레임워크 버전과 동일한 XGBoost 프로세서를 사용합니다. 이에 따라 제공된 평가 스크립트를 실행하는 데 사용되는 컨테이너가 결정됩니다.

다음으로 output 파라미터를 사용하여 출력 디렉터리를 제공합니다. 이 경로는 Processing 작업 인스턴스의 로컬 경로입니다. Processing Job Output의 내용은 JSON 파일로 저장하여 Amazon Simple Storage Service (Amazon S3) 에 출력으로 저장됩니다.

그런 다음 property_files 파라미터를 사용하여 속성 파일을 설정합니다. 이 속성 파일은 이 단계와 조건 단계 사이의 모든 값이나 메트릭을 전달하는 데 사용됩니다. 속성 파일은 Processing 작업의 일부로 위의 출력 디렉터리에 기록되어야 합니다.

마지막으로 SageMaker Pipeline에 평가 단계를 위한 ProcessingStep을 생성합니다. 이는 훈련 단계에 의존성을 설정하기 위해 depends_on을 설정합니다. 그리고 평가 결과 파일을 속성으로 추가합니다.

def _step_evaluation(self, ):
        
        strInstanceType = "ml.m5.2xlarge"
        nInstanceCount = 1
        
        strProcPrefixPath = "/opt/ml/processing"
        strTrainDataPath = os.path.join(self.strDataPath, 'train', "train.csv")
        strTestDataPath = os.path.join(self.strDataPath, 'test', "test.csv")
        
        strOutputPath = os.path.join(
            "s3://{}".format(self.strBucketName),
            self.strPipelineName,
            "evaluation",
            "output"
        )

        strCodeLocation = os.path.join(
            "s3://{}".format(self.strBucketName),
            self.strPipelineName,
            "evaluation",
            "backup_codes"
        )
        
        evaluation_processor = FrameworkProcessor(
            estimator_cls=XGBoost,
            framework_version="1.7-1",
            image_uri=None,
            role=self.strExecutionRole,
            instance_type=strInstanceType,
            instance_count=nInstanceCount,
            base_job_name="evaluation",             
            sagemaker_session=self.pipeline_session
        )
        
        step_evaluation_args = evaluation_processor.run(
            code=f'evaluation.py',
            source_dir=f'{self.strSource}/source/evaluation/sourcedir.tar.gz',
            inputs=[
                ProcessingInput(
                    source=strTrainDataPath,
                    input_name="train_data",
                    destination=os.path.join(strProcPrefixPath, "train")
                ),
                ProcessingInput(
                    source=strTestDataPath,
                    input_name="test_data",
                    destination=os.path.join(strProcPrefixPath, "test")
                ),
                ProcessingInput(
                    source=self.training_process.properties.ModelArtifacts.S3ModelArtifacts,
                    input_name="model_weight",
                    destination=os.path.join(strProcPrefixPath, "model")
                )
            ],
            outputs=[
                ProcessingOutput(
                    source=os.path.join(strProcPrefixPath, "output"),
                    output_name='evaluation',
                    destination=strOutputPath,
                )
            ],
        )
        
        self.evaluation_report = PropertyFile(
            name="EvaluationReport",
            output_name="evaluation", 
            path="evaluation.json",
        )
        
        self.evaluation_process = ProcessingStep(
            name="EvaluationProcess",
            step_args=step_evaluation_args,
            depends_on=[self.training_process],
            property_files=[self.evaluation_report],
            cache_config=self.cache_config,
            retry_policies=self.retry_policies
        )
Python

다음과 같은 과정으로 ML모델의 평가 과정을 자동화하고, 파이프라인의 일부로 통합하여 모델 성능을 객관적으로 측정할 수 있습니다. 이를 통해 모델의 품질을 보장하고, 지속적인 모델 개선 프로세스를 지원합니다.

모델 저장

이렇게 학습된 모델은 Amazon SageMaker Model Registry에 저장하여 버전 관리를 수행했습니다. 이렇게 저장된 모델은 새로운 데이터로 업데이트할 수 있도록 관리되었습니다. 만약 기존에 SageMaker Endpoint에 배포된 모델과의 성능 비교 Evaluation을 통해 성능 요구사항에 미치지 못할 경우 재학습하도록 하였습니다.

6. 모델 배포 추론 단계

아래는 ML모델 배포 및 추론 과정 샘플 코드입니다. 이를 통해 ML모델 배포 과정을 자동화하고, 파이프라인의 일부로 통합하여 모델을 실제 서비스에 적용하는 과정을 간소화합니다.

배포 설정에서 인스턴스 타입, 개수, Endpoint 이름 등을 설정합니다. 모델이 매월 생성되기 때문에 Endpoint 이름에 날짜 정보를 결합합니다. 다음으로 FrameworkProcessor에서는 학습, 평가와 마찬가지로 XGBoost 프레임워크를 사용하여 배포 작업을 수행할 프로세서를 설정합니다.

step_deploy_args에서는 배포작업을 설정합니다. 배포과정에서 필요한 inference.py 스크립트와 requirements.txt 파일을 포함합니다. 그리고 훈련된 모델 아티팩트를 사용하여 새로운 엔드포인트를 생성합니다.

특히, 모델 배포는 배포 개발인력이 부족한 팀에게 큰 도전 과제였습니다. SageMaker Endpoint를 통해 실시간으로 발생하는 호출 데이터에 대한 성공 확률을 예측할 수 있었으며, 보다 정확하고 빠른 배차가 가능하게 되었습니다. 티머니 온다택시 서비스 안정성을 위해서는 모델 지연시간이 100ms를 넘지 않는 것을 목표로 합니다. 따라서 기준 지연시간을 초과할 경우 Endpoint에 Auto Scaling을 적용하였고, 급격히 증가하는 호출 요청에도 안정적으로 대응할 수 있었습니다.

def _step_deploy(self, ):
        
        strInstanceType = "ml.m5.2xlarge"
        nInstanceCount = 1

        ver_date = today.strftime('%Y-%m-%d')
        strEndpointName = "Endpoin-Name" + ver_date
        strProcPrefixPath = "/opt/ml/processing"
        
        deploy_processor = FrameworkProcessor(
            estimator_cls=XGBoost,
            framework_version="1.7-1",
            image_uri=None,
            role=self.strExecutionRole,
            instance_type=strInstanceType,
            instance_count=nInstanceCount,
            base_job_name="deploy",
            sagemaker_session=self.pipeline_session
        )
        
        step_deploy_args = deploy_processor.run(
            code=f'deploy.py',
            source_dir=f'{self.strSource}/source/deploy/sourcedir.tar.gz',
            inputs=[
                ProcessingInput(
                    source=f'{self.strSource}/source/deploy/inference.py',
                    input_name="inference-py",
                    destination=os.path.join(strProcPrefixPath, "inference")
                ),
                ProcessingInput(
                    source=f'{self.strSource}/source/deploy/requirements.txt',
                    input_name="requirements-txt",
                    destination=os.path.join(strProcPrefixPath, "requirements")
                ),
            ],
            arguments=[
                "--prefix_deploy", strProcPrefixPath, \
                "--region", self.strRegionName, \
                "--instance_type", strInstanceType, \
                "--model_data", self.training_process.properties.ModelArtifacts.S3ModelArtifacts, \
                "--endpoint_name", strEndpointName, \
                "--execution_role", self.strExecutionRole, \
            ],
            job_name="deploy",
        )
        
        self.pm.put_params(key=self.strPrefix + "-ENDPOINT-NAME", value=strEndpointName, overwrite=True)
        
        self.deploy_process = ProcessingStep(
            name="DeployProcess",
            step_args=step_deploy_args,
            depends_on=[self.evaluation_process],
            cache_config=self.cache_config,
            retry_policies=self.retry_policies
        )        
Python

7. MLOps 구성 및 자동화

MLOps는 Amazon SageMaker Pipelines를 사용해 구성되었습니다. 이 파이프라인은 데이터 전처리, 모델 훈련, 평가, 등록, 배포의 모든 단계를 자동으로 실행하며, 각 단계 간의 종속성을 설정하여 전체 ML 워크플로우가 원활하게 이루어지도록 했습니다.

티머니는 매월 학습, 참조 데이터가 주기적으로 변경되는 상황에서 학습 데이터 전처리를 위한 ‘실행파일.py’도 주기적으로 변경되기 때문에 이를 엔드 투 엔드로 자동화하는 방향을 검토했습니다.

다양한 방법을 검토한 끝에 2023년에 출시한 SageMaker Notebook Job을 적용하여 문제를 해결하였습니다. JupyterLab 내 Notebook Job extension을 활용하면 Scheduling Job을 별도의 Amazon Event Bridge 또는 AWS Lambda 함수를 구성하지 않고도 손쉽게 구성할 수 있습니다.

그림4. 티머니 온다택시의 MLOps 자동화 적용하여 개선된 아키텍처

이를 통해 이벤트 기반으로 특정 작업을 트리거하여 필요한 시점에 자동으로 처리될 수 있도록 했습니다. 이러한 자동화는 티머니가 모델 성능을 지속적으로 개선하고, 새로운 데이터를 반영하는 과정을 효율적으로 관리할 수 있게 해주었습니다.

아래와 같이 매월 특정 날짜, 특정 시간에 Cron job을 수행하도록 Notebook Job을 구성했습니다.

그림5. SageMaker Notebook Job 구성 예제

구현 결과 : 실시간 택시 호출의 ML모델 활용한 마일리지 결정

아래와 같이 고객이 온다택시를 호출했을 경우, 기사 앱에서는 ML의 결정에 따라 해당 콜의 마일리지 지급 여부와 얼마의 마일리지가 지급되는 지 실시간으로 확인할 수 있습니다.

이에 대한 결정은 수요과 공급을 대표하는 변수들을 바탕으로 ML이 호출별 배차성공 확률을 계산하고, 해당 확률이 내부적으로 정한 기준보다 낮으면 지급으로 결정하게 됩니다.

그림6. 티머니 온다택시 AI/ML 적용 후 기사 앱 화면 예시

아울러 다음과 같은 부가적인 효과도 있었습니다:

  • 모델의 실시간 예측 결과에 따른 마일리지 지급으로 비용 효율성을 극대화할 수 있었습니다. 성공 확률이 낮은 호출에만 기사 마일리지를 지급하여 기존에 불필요하게 지급되던 비용을 효율적으로 사용할 수 있었습니다.
  • 배차의 성공률 또한 증가시킬 수 있었습니다. 성공확률이 낮은 호출에 기사 마일리지를 지급함으로써 기사님의 수락율을 높이고, 이에 따라 호출의 배차 성공률이 증가했습니다.
  • SageMaker Endpoint를 통한 호출의 성공확률 실시간 예측 후 마일리지 지급 자동화는 기존에 수동으로 작업하는 인원의 인건비 감소 효과도 창출했습니다. 기존에는 특별한 상황이 있는 경우 각 부서 담당자가 마일리지 지급을 결정하여 시스템에 반영했지만, 모델에 의한 성공확률 예측으로 특별한 상황을 모델 스스로 인식하고, 자동으로 마일리지를 지급하면서 수작업을 없애는데 기여했습니다.
  • ML모델의 주기적인 자동 업데이트를 통해 수요와 공급의 변화를 반영하여 예측 정확도를 유지할 수 있었습니다. 수요와 공급의 데이터 분포 변화는 예측 정확도에 영향을 미치는 중요한 부분이기 때문에, 주기적인 업데이트로 데이터의 분포 변화를 놓치지 않고 캐치하여 모델의 안정적인 성능을 가능하게 했습니다.

마무리

이번 블로그에서는 티머니의 스마트 호출 배차를 위해 온프레미스에서 개발한 ML모델을 Amazon SageMaker로 마이그레이션하여 보다 효율적인 MLOps환경을 구축한 과정을 소개했습니다. 이 솔루션을 통해 택시 호출 성공확률을 높여서 빠른 배차를 지원하면서도 서비스 운영비용을 최적화할 수 있었고, MLOps를 도입하여 적은 인원으로도 반복적인 작업을 생산성있게 운영할 수 있게 되었습니다.

온프레미스의 모델을 SageMaker Endpoint로 배포하는 방법은 BYOM 예제 코드를 참고 하시면 되며, AWS 공식 GitHub의 amazon-sagemaker-examples/ml_ops 에서 더 많은 예시 코드를 확인할 수 있습니다. 아울러, SageMaker를 더 잘 활용하기 위해 aws-samplesAWS AIML 한글워크샵 모음 공식 GitHub을 방문 해보시길 바랍니다.

Yongju Ban

Yongju Ban

티머니 반용주 매니저는 비즈니스의 다양한 문제들을 데이터를 기반으로 해결하려고 노력합니다. 문제의 종류에 따라 Machine learning 모델을 개발하거나 데이터 분석 결과를 제공하여 데이터 기반의 의사결정을 지원합니다.

Hyunseo Gu

Hyunseo Gu

티머니 구현서 매니저는 티머니 빅데이터팀 소속으로, 다양한 비즈니스 문제들을 발굴하고, 데이터 분석에 기반해 문제를 해결하는 역할을 하고 있습니다.

Jihoon Oh

Jihoon Oh

티머니 오지훈 매니저는 비즈니스 로직 개발부터 AWS 아키텍처 설계까지 다양한 기술적 과제를 해결하며 서비스의 안정성과 확장성을 지원합니다. 또한, 컴플라이언스 요구사항을 충족하기 위해 시스템을 설계하고 운영하며, 비즈니스 목표 달성을 위한 기술적 기반을 제공합니다.

Byungho Lee

Byungho Lee

이병호 솔루션즈 아키텍트는 네트워크 엔지니어링 분야 전문성을 갖고 엔터프라이즈 고객을 위해 일하고 있습니다. 복잡한 요구사항을 해결하기 위한 최적의 솔루션을 설계하고 지속적으로 클라우드 이니셔티브를 확대하고자 노력하고 있습니다.

Hyeonsang Jeon

Hyeonsang Jeon

전현상 솔루션즈 아키텍트는 데이터 기반 멀티테넌트 AI/ML 플랫폼의 핵심 소프트웨어 개발자로 활동하였으며 통신부터 제조 및 금융에 이르기까지 다양한 SW 및 데이터 AI/ML분야에 기여했습니다. AI/ML 소프트웨어 엔지니어링 경험은 현재 AWS 솔루션즈 아키텍트의 역할에 보탬이 되고 있습니다. 금융 고객 및 AI/ML 도입을 원하는 기업들에게 AWS 플랫폼을 최적화하여 활용하는 방법을 제공하며, AI/ML 트렌드의 확산에 적극적으로 기여하고 있습니다.