Project Detail

Monkey Travel

Monkey Travel — Commerce Platform

해외 호텔 공급사들의 방대한 데이터를 처리하는 병렬 파이프라인을 구축하고, 운영 리소스를 획기적으로 줄여 '기술이 비즈니스 이익을 만드는' 사례를 증명했습니다.

배치 수행 시간 23일 → 3.8일 (83% 단축)
글로벌 OTA 데이터 ETL 완전 자동화
운영 인력 비용 절감 및 업무 효율 극대화

글로벌 OTA(Agoda, Expedia, Hotelbeds, FITRuums 등)로부터 전 세계 호텔 데이터를 수집/적재하는 파이프라인을 설계하여 데이터 갱신 속도를 23일에서 3.8일로 단축(83% 성능 향상)했습니다. 수작업에 의존하던 운영 프로세스를 100% 자동화하여 비용 절감과 비즈니스 확장성을 확보했습니다.

PIPELINE

OTA 데이터 파이프라인

Agoda, Expedia, Hotelbeds, FITRuums 배치를 Airflow와 PHP 워커로 자동화하고, 실패 시 재시도, 재개 지점을 관리합니다.

Airflow, PHP Batch, Resilient DAGs

DATA

데이터 정규화 & 관리 UX

대륙-국가-도시-호텔-객실 계층을 정규화하고 어드민에서 신규 데이터 생성, 검증 플로우를 개선했습니다.

MariaDB, Redis, OpenSearch

OPS

운영 효율화 & 피드백

현지 운영팀이 업데이트된 재고를 즉시 파악할 수 있도록 알림, 하이라이트를 제공하고, 배치 상태를 대시보드로 가시화했습니다.

Alerting, Dashboard, Telegram Bot

Monkey Travel 프론트오피스 — 호텔 대시보드

1 / 20

프로젝트 개요

Monkey Travel은 동남아시아 지역을 중심으로 호텔, 골프, 투어, 스파 등 다양한 여행 상품을 예약할 수 있는 플랫폼입니다. 이번 프로젝트는 기존의 수작업 위주의 예약 관리 프로세스를 자동화하고, 고객 경험을 대폭 개선하기 위해 진행되었습니다. 특히, 데이터 정규화 및 동기화를 통해 비효율성을 제거하고, 유지관리 비용을 획기적으로 절감하는 데 성공하였습니다.

주요 특징

호텔 데이터 공급업체 OTA(Agoda, Expedia, Hotelbeds, Fitruums)로부터 배치 프로세스를 통해 데이터를 수집, 가공하여 DB에 영속화
Airflow를 활용한 호텔 데이터 ETL 프로세스 스케쥴링 및 자동화
Redis와 OpenSearch를 이용한 빠른 데이터 조회
기존의 몽키트래블 어드민 보다 나은 UI/UX를 제공하여 관리 환경 개선
업데이트된 호텔 혹은 객실에 대해 현지 어드민에게 피드백 제공

주요 책임 및 성과

대륙, 국가, 도시, 도시 내 지역, 호텔, 객실, 시설, 어메니티, 액티비티 등 다양한 데이터 모델 설계 및 정규화
OTA 벤더별 1차 메타데이터 테이블, 2차 합성 테이블, 3차 최종 마이그레이션 테이블 구성 등 테이블 설계
OTA 벤더별 API 분석, 활용 방안, 배치 구동 절차, 시퀀스 플로우 등 문서 작성
호텔 데이터 공급업체 OTA(Agoda, Expedia, Fitruums, Hotelbeds)로 부터 전 세계 호텔 데이터 수집/가공/적재하는 ETL PHP 배치 프로세스 개발 및 Airflow DAG 작성을 통한 일간/주간 전 세계 호텔 데이터 적재 배치 프로세스 자동화
Agoda 배치 실행 시간 83% 단축 (기존 23일 → 3.8일): Airflow Task 병렬 분할 처리 도입
Expedia 배치 실행 시간 75% 단축 (기존 2일 → 0.5일)
새로운 호텔 공급업체 Hotelbeds, FITRuums ETL 설계 및 구현
호텔, 객실, 호텔 상세 정보 생성/관리를 위한 백오피스 비즈니스 개발
퍼블리싱 된 프론트오피스 비즈니스 로직 개발

기술 스택

Backend: PHP 8.3, Smarty, Python
Database: MariaDB, Redis, OpenSearch
Infra: Docker, Amazon EC2
Batch Scheduling: Airflow

담당 역할

호텔, 객실과 관련된 주요 데이터 전반에 걸쳐 담당했습니다. 모든 주요 데이터를 최초 클린 상태에서 배치 프로세스를 통해 전 세계 호텔 데이터를 한 번 쌓은 후 이후부터 Airflow DAG를 통해 일간/주간 단위로 스케쥴링하여 지속해서 호출해 기존의 데이터를 업데이트하도록 했습니다. 진행하면서 ETL 프로세스 실행 시간 최적화와 자동화를 성공적으로 수행하였습니다.

또한, 생성한 호텔 데이터를 통해 객실, 시설, 어메니티, 액티비티 등 다양한 데이터 모델을 설계하고, 이를 정규화하여 어드민에서 이를 관리할 수 있도록 구현하였습니다. 이를 통해 어드민이 기존 데이터를 기반으로 새로운 호텔과 객실 데이터를 새로 생성하고, 이는 어느 OTA에서 오는 데이터인지 쉽게 인지할 수 있도록 했습니다.

배운 점

배치 프로세스 완전 자동화(+에러에 대한 내성)에 대한 경험

이 프로젝트를 통해 데이터 모델링과 ETL 프로세스에 대한 이해를 높일 수 있었습니다. 특히, 데이터 정규화를 통해 데이터의 중복성을 제거하고, 데이터의 일관성을 유지하는 방법에 대해 배울 수 있었습니다.

또한, Airflow를 활용한 ETL 프로세스를 자동화 하고, 수동으로 진행할 때의 실수를 줄일 수 있다는 점을 배웠습니다.

호텔 데이터 적재 배치를 자동화 하면서 정말 엄청나게 성취감이 느껴지는 부분이 있었는데, 기존의 몽키트래블은 태국 본사의 현지인 약 4~50명의 어드민이 직접 API를 하나 하나씩 수동으로 호출해 호텔/객실 데이터가 업데이트가 필요한지 확인하고 필요한 경우 업데이트를 수행하는 엄청나게 비효율적인 방식이였습니다. 이번 ETL 자동화 작업을 통해 기존의 수작업 위주의 데이터 관리 방식에서 탈피해 유지보수 비용을 획기적으로 절감하는 데 성공하였습니다. 개발자는 비즈니스 문제를 해결하는 해결사가 되어야 한다는 말을 유튜브에서 본 적이 있는데, 이 순간 만큼은 그 말이 정말 와닿았던 순간이었습니다. 우리의 노력으로 다른 이익 집단 혹은 관계자가 시간적, 비용적 이점을 아주 크게 얻을 수 있었기 때문인 것 같습니다.

병목을 발생시키는 또 다른 케이스의 발견 및 적절한 대응

그 다음으로 배운 점으로는 배치 프로세스의 실행 시간에 악영향을 끼치는 요인으로는 코드의 시간 복잡도나 RDB만 있는 것이 아니다. 였습니다. Expedia, Hotelbeds, Fitruums의 경우 전 세계 호텔 데이터를 제공할 때 jsonl 파일 하나로 내려주기에 이 파일을 읽어서 전체 호텔 데이터를 업데이트 처리하는데 그리 많은 시간이 걸리지 않았습니다. 반면 Agoda의 경우는 REST API로 호텔 데이터를 하나 하나씩 단건으로 호출해 적재해야 했습니다.

이 경우 외부 API 호출 시간이 가장 큰 병목이 됩니다. 심지어 Rate Limit도 엄격하게 걸려 있어 HTTP Response로 429(Too Many Requests) 응답이 간헐적으로 내려오기도 해 로직에 Back off 전략을 적용해 429 응답이 내려온 경우 대기하는 시간을 늘려가며 데이터를 수집해야 했습니다. 이 과정에서 Agoda 주간 배치의 실행 시간을 계산해보니 총 23일이 걸렸습니다. 주간 배치임에도 불구하고 이는 너무나도 긴 시간이었습니다. 따라서 이를 해결하기 위해 Airflow DAG에서 다음의 방식을 고안해 냈습니다.

병렬 분할 처리 방식

전체 호텔의 마지막 ID를 조회합니다.
마지막 ID를 기준으로 분할 요인 수(split_number)를 기준으로 나누어 구간을 구합니다.

1# last_record_id 를 split_number 로 범위 분할 함수
2def split_into_ranges(last_record_id, split_number):
3    split_count = last_record_id // split_number
4    ranges = []
5    for i in range(split_number):
6        start_id = i * split_count if i == 0 else ranges[-1]['end_id']
7        end_id = (i + 1) * split_count if i < split_number - 1 else last_record_id
8        ranges.append({'start_id': start_id, 'end_id': end_id})
9    return ranges
10
11# 생성된 태스크 예시
12# split_number = 10
13# Last Hotel Id = 1,000,000
14# 생성된 각 구간 => 
15# [
16#     {start_id: 0, end_id: 100000}, 
17#     {start_id: 100000, end_id: 200000},
18#     ..., 
19#     {start_id: 900000, end_id: 1000000}
20# ]

만약 split_number가 10이고, 마지막 호텔 ID가 1,000,000이라면, 1백만을 10개의 구간으로 나누어 생성합니다.

각 구간별로 태스크를 생성해 병렬로 실행합니다. 각 태스크는 Agoda API를 호출해 호텔 데이터를 수집하고, 이를 DB에 적재하는 작업을 수행합니다.

1# 호텔 디테일 데이터 업데이트 병렬 수행
2hotel_detail_tasks = extract_and_load_hotel_details.partial(
3    batchRunId=batch_run_id,
4    batchRecordId=batch_record_id,
5    currTaskStep=curr_task_step
6).expand(task_params=hotel_detail_task_params)
7
8# Step 3: 호텔 디테일 데이터 업데이트 병렬 수행
9@task(on_failure_callback=stop_dag, trigger_rule='none_failed')
10def extract_and_load_hotel_details(task_params, batchRunId, batchRecordId, currTaskStep):
11    if currTaskStep > CURRENT_STEP_NUMBER:
12        logging.info(f"Step {CURRENT_STEP_NUMBER} 이미 완료되었으므로 스킵합니다.")
13        return []
14
15    try:
16        ...생략...
17
18        # 호텔 디테일 데이터 업데이트 호출
19        call_endpoint(
20            supplier='otaBatchAgoda',
21            batch_type=endpoint,
22            request_body={
23                'id': task_status_id,  # 현재 실행 중인 태스크의 레코드 ID를 함께 전달
24                'language': language,
25                'start_id': start_id,
26                'end_id': end_id,
27            },
28        )
29
30        # 태스크 상태 업데이트: 성공
31        update_task_status(task_status_id, TASK_STATUS_SUCCESS)
32
33        logging.info(f"[Step 3] 호텔 디테일 데이터 업데이트 병렬 수행 단계를 완료했습니다.")
34    except Exception as e:
35        # 태스크 상태 업데이트: 실패
36        update_task_status(task_status_id, TASK_STATUS_FAILED, str(e))
37        logging.error(f"[Step 3] 에러 발생. 메시지: {str(e)}")
38        raise

이 start_id와 end_id를 전달받은 PHP 백엔드 애플리케이션 API 중 하나는 다음과 같습니다.

1while (true) {
2	if ($currentCursor > $endRecordId) {
3	    $this->logger->info("[SUCCESS] 업데이트 처리가 필요한 호텔이 존재하지 않습니다. Hotel Full Info(FEED_ID:19) 적재 배치를 정상적으로 종료합니다. 진행된 배치 범위({$startRecordId}~{$endRecordId})");
4	    return;
5	}
6	
7	// [A] 임계값 단위로 현재 태스크 레코드의 task_param의 start_id 업데이트
8	if ($taskUpdateCounter++ >= $this->TASK_WEEKLY_HOTEL_DETAILS_UPDATE_COUNT_THRESHOLD) { // start_id 업데이트
9	    $taskParams['hotel_id_range']['start_id'] = $currentCursor;
10	    $this->otaBatch->updateTaskParams($taskRecordId, json_encode($taskParams));
11	    $taskUpdateCounter = 0;
12	}
13	
14	$hotelsArray = $this->findIdAndHotelIdByIdGreaterThan($currentCursor, $languageCode, $this->HOTEL_CHUNK_SIZE);
15	
16	if (empty($hotelsArray)) {
17	    $this->logger->debug("[SUCCESS] 업데이트 처리가 필요한 호텔 목록이 존재하지 않습니다. Hotel Full Info(FEED_ID:19) 적재 배치를 정상적으로 종료합니다.");
18	    return;
19	}
20	
21	foreach ($hotelsArray as $hotel) {
22	    $hotelFullInfo = BatchAgodaHotel::fromHotelFullInformationJsonForUpdate(
23	        $this->callContentAPI(ContentAPIRequest::toHotelInfoQueryParams($this->FEED_ID_HOTEL_FULL_INFO, $languageCode, $hotel['hotelId'])),
24	        $hotel['id']);
25	
26	    $this->arrayUpdater(
27	        $this->TABLE_PRODUCT_HOTEL_SUPPLIER_AGODA_HOTEL . $languageCode->label() . '.hotel',
28	        [$this->convertObjectToArray($hotelFullInfo)]);
29	
30	    $this->logger->info("Hotel ID({$hotel['hotelId']}) Hotel Full Info 조회 API 호출 및 DB 적재 완료.");
31	
32	    // 짧은 간격으로 잦은 API 호출을 방지하기 위해 대기
33	    sleep($this->API_CALL_INTERVAL);
34	
35	    unset($hotelFullInfo);
36	}
37	
38	$this->logger->debug('Current Cursor: ' . $currentCursor . ' | Next Cursor: ' . end($hotelsArray)['hotelId']);
39	// 현재 hotelId 배치 커서를 다음 포인트로 이동
40	$currentCursor = end($hotelsArray)['id'];
41	
42	unset($hotelsArray);
43	}
44}

주석 [A]를 보시면 업데이트 카운트가 Threshold에 도달할 때마다 즉, 호텔을 특정 개수만큼 업데이트 처리한 경우 start_id를 현재 Cursor(Hotel ID)로 업데이트합니다. 만약 서버가 다운된 경우 Airflow가 자체적으로 Failover 동작으로 DAG를 자동으로 다시 실행하고, 기존 실행에 실패했던 DAG의 각 태스크의 중단점(start_id)부터 end_id까지 진행하게 됩니다. 이러한 방어장치 없이 선형적인 방식에서는 Agoda 호텔 데이터 적재 시간에 총 23일이 필요했고, 에러가 발생하면 다시 처음부터 실행해야 했습니다.

하지만 이러한 안전장치를 도입해 긴 작업 도중 실패하더라도 다시 중단점 부터 실행할 수 있도록 하였으며, 더 나아가 병렬 처리를 통해 Agoda 호텔 데이터 적재 시간을 3.8일로 단축할 수 있었습니다.

※ 23일이 산정된 이유는 단건 API 호출 후 다음 API 호출까지 대기하는 시간이 2초이고(Rate Limit 피하기 위함), Agoda에는 총 1,000,000개의 호텔 데이터가 있기 때문에 1,000,000 * 2초 = 23일로 산정된 것입니다. 이는 백오프 대기 시간을 고려하지 않았기 때문에 실제 시간보다 더 오래 걸릴 수도 있습니다.

DAG 실행 상태 관리 테이블

DAG에서 현재 실행 중인 각 태스크의 start_id, end_id 관리 테이블 정의는 다음과 같습니다.

1create table dag_execution_status
2(
3    ...생략...
4    otaId             TINYINT(1) UNSIGNED                                                         NOT NULL COMMENT 'OTA ID',
5    batchType         ENUM('DAILY', 'WEEKLY')                                                     NOT NULL COMMENT '배치 타입 (일간/ 주간)',
6    batchRunStatus    ENUM('STANDBY', 'RUNNING', 'SUCCESS', 'FAILED') DEFAULT 'STANDBY'           NOT NULL COMMENT '배치 진행 상태',
7    currTaskStep      INT                                             DEFAULT 0                   NOT NULL COMMENT '현재 진행 중인 태스크 단계',
8    taskParams        LONGTEXT                                        DEFAULT '{}'                NOT NULL COMMENT '전체 배치에 필요한 공통 파라미터',
9    errorMessage      TEXT                                            DEFAULT ''                  NOT NULL COMMENT '에러 내용',
10    lastHeartbeat     DATETIME                                                                    NULL COMMENT '마지막 하트비트 시간',
11    ...생략...
12);
13
14
15create table dag_task_execution_status
16(
17    ...생략...
18    dagExecutionStatusId int                                                                                             NOT NULL COMMENT '참조하는 배치 실행 상태 ID',
19    taskName                      VARCHAR(255)                                                                           NOT NULL COMMENT '태스크 이름',
20    language                      VARCHAR(10)                                                                            NOT NULL COMMENT '언어 코드 (예: en, ko)',
21    taskStatus                    ENUM ('QUEUED', 'RUNNING', 'SUCCESS', 'FAILED', 'SKIPPED') DEFAULT 'QUEUED'            NOT NULL COMMENT '태스크 상태',
22    startId                       BIGINT                                                                                 NULL COMMENT '처리 시작 ID',
23    endId                         BIGINT                                                                                 NULL COMMENT '처리 종료 ID',
24    taskParams                    LONGTEXT                                                   DEFAULT '{}'                NOT NULL COMMENT '태스크에 전달된 파라미터 목록',
25    errorMessage                  TEXT                                                       DEFAULT ''                  NOT NULL COMMENT '에러 내용',
26    ...생략...
27);

dag_execution_status 테이블이 DAG 전반적인 실행 상태를 다루고, dag_task_execution_status 테이블이 병렬로 처리 중인 각 태스크의 실행 상태를 다룹니다. dag_task_execution_status.taskStatus는 태스크의 상태를 나타내며, startId, endId가 각 태스크의 실행 범위를 나타냅니다. 이 값을 통해 실패한 DAG가 새로 실행되면 FAILED 상태의 태스크를 찾고, 중단된 startId, endId 부터 진행하도록 합니다.

그런데 이러한 일련의 Failover 동작은 서버가 셧다운 되었을 때 실행된다고 말씀드렸는데요, 서버가 모종의 이유로 강제로 종료되었을 때 이를 알아차릴 방법이 필요했습니다. 따라서 서버 측에서 배치 프로세스가 정상적으로 진행 중일 경우 dag_execution_status.heartLastBeat를 지속해서 갱신하고, DAG에서 일정 주기로 서버에 healthy API를 요청해 heartLastBeat가 지속해서 업데이트 되고 있는지 체크하도록 했습니다. 만약 이 값이 현재 시간으로부터 약 35분 동안 업데이트되지 않은 경우 서버가 비정상적으로 다운된 것으로 인지하고 DAG를 다시 실행하도록 했습니다.

다양한 일간 배치 케이스의 대응

Agoda, Expedia, Hotelbeds, Fitruums 모두 호텔 데이터 적재 배치를 만들 때, 주간 배치 뿐만 아니라 일간 배치 또한 만드는 것을 권유 합니다. 주간 배치는 각 벤더가 제공하는 모든 호텔의 메타데이터 업데이트이고, 일간 배치는 호텔이 활성화, 비활성화, 업데이트 되었는지 최근 변동 사항을 업데이트하는 과정입니다. 호텔의 공지사항 혹은 객실 변동 사항이 매우 잦으므로 정합성을 빠르게 맞출 필요가 있기 때문에 일간 배치를 거의 필수적으로 개발해야만 합니다. 앞서 말씀드린 활성화/비활성화/업데이트 세 가지의 케이스에 대해 각 벤더가 제공하는 REST API를 통해 구현해야 합니다. 이를 위해 OTA 벤더 중립적인 일간 배치 테이블을 생성했습니다.

1create table ota_daily_batch_execution_status
2(
3    ... 생략...
4    
5    mTypeId           TINYINT UNSIGNED
6         NOT NULL COMMENT '검색 타입 (1: 업데이트 된 호텔 목록, 2: 새로 활성화 된 호텔 목록, 3: 비활성화/ 폐쇄 된 호텔 목록)',
7    mDate             DATETIME
8         NOT NULL COMMENT '검색 기준 일시',
9    status            ENUM ('EXTRACTING', 'EXTRACTED', 'TRANSFORMING', 'TRANSFORMED', 'LOADING', 'LOADED', 'SUCCESS', 'FAILED') DEFAULT 'EXTRACTING'
10         NOT NULL COMMENT 'BATCH 진행 상태',
11    hotelIds          LONGTEXT                                         DEFAULT '[]'
12         NOT NULL COMMENT '업데이트가 필요한 호텔 ID 목록'
13         CHECK (JSON_VALID(`hotelIds`)),
14    updatedIds        LONGTEXT                                         DEFAULT '[]'
15         NOT NULL COMMENT '업데이트에 성공한 호텔 ID 목록 (전 세계 대상)'
16         CHECK (JSON_VALID(`updatedIds`)),
17    failedId          BIGINT
18         NULL COMMENT '업데이트에 실패한 호텔 ID 목록',
19    failedMessage     TEXT                                             DEFAULT ''
20         NOT NULL,
21    isNotiPublished   TINYINT(1) UNSIGNED                              DEFAULT 0
22         NOT NULL,
23    lastJobStep       TINYINT(2) UNSIGNED                              DEFAULT 0
24         NOT NULL COMMENT '배치 실행 단계 (0. NONE (시작 전 단계), 1. ORIGIN_FULL_INFO, 2. ORIGIN_OTHER_INFO, 3. ORIGIN_LOCAL_INFO, 4. ORIGIN_FACILITY_INFO, 5. SUPPLIER_MASTER, 6. SUPPLIER_FACILITY, 7. SUPPLIER_ROOM, 8. MASTER_HOTEL, 9. MASTER_FACILITY, 10. MASTER_ROOM)',
25    syncDate          DATETIME
26         NULL COMMENT 'Supplier Master -> Origin Master 테이블로의 동기화 일시'
27    
28    ... 생략...
29) comment 'OTA 벤더 중립적인 일간 배치 테이블. 업데이트/활성화/비활성화 된 호텔 Ids 저장';

테이블 필드 설명

mTypeId: 업데이트/활성화/비활성화 유형 중 어느 유형으로 배치가 진행 중인지 알기 위한 타입 정보입니다. OTA 벤더들은 공통적으로 이 세 가지 타입의 일간 데이터를 제공합니다.
mDate: 검색 대상 날짜 구간을 의미합니다.
status: 배치 진행 상태를 의미합니다. 관리 주체는 Airflow DAG 입니다.
hotelIds: 배치 EXTRACT 단계에서 업데이트가 필요한 호텔 ID 목록을 저장합니다.
updatedIds: 업데이트를 정상적으로 마친 호텔 ID 목록을 저장합니다.
failedId: 업데이트에 실패한 호텔 ID를 저장합니다.
failedMessage: 업데이트를 실패한 이유를 저장합니다.
isNotiPublished: 몽키트래블 어드민에게 알림이 전송됐는지에 관한 여부입니다. 어드민 페이지의 호텔 리스트에서 업데이트 된 호텔에 하이라이트 표기하고, 텔레그램으로 업데이트 알림을 보냅니다.
lastJobStep: 마지막으로 수행된 Step이 어디인지 알기 위한 기록입니다. 문제가 발생해 배치가 종료된 경우 이 Step 부터 진행하도록 해 시간적, 비용적 리소스를 줄입니다.
syncDate: 어드민 페이지에서 실제로 어드민이 업데이트를 진행한 날짜입니다. 최종 업데이트 결정권은 몽키트래블 어드민이 가지며, 어느 어드민이 업데이트를 수행했는지 알 수 있습니다.

Agoda 일간 호텔 ID 적재 배치 로직

OTA 벤더 중 Agoda의 일간 호텔 ID 적재 배치 로직은 다음과 같습니다.

1/**
2 * [Extract] FEED_ID 32 mtypeid(1 or 2 or 3): 업데이트 된 호텔 목록 조회 & 적재
3 *
4 * @param int $mtypeid
5 * @param string $mdate
6 * @return mixed|void
7 * @throws \Exception
8 */
9public function extractUpdatedHotelIdsJob(int $mtypeid, string $mdate)
10{
11    try {
12        $this->logger->info("FEED_ID 32 mtypeid(1 or 2 or 3): 업데이트 된 호텔 목록 조회 & 적재 작업을 시작합니다. 파라미터 (mtypeid: $mtypeid, mdate: $mdate)");
13
14        /*
15         * [A]
16         * 초입 단계에서 ota_daily_batch_execution_status 테이블에
17         * 입력으로 들어온 mtypeid, mdate가 일치하는 레코드가 존재하고 &
18         * status가 RUNNING 인 경우 Extract 프로세스 조기 종료
19         */
20        $existingRunningRecord = $this->findTodayRunningRecordByMtypeid($mtypeid);
21
22        // RUNNING 레코드가 있는 경우 배치 프로세스 조기 정상 종료
23        if (!empty($existingRunningRecord)) {
24            $this->logger->info(BatchErrorCode::AGODA_DAILY_IS_RUNNING->getMessage("ko"));
25            return [
26                "result" => false,
27                "data" => ["code" => BatchErrorCode::AGODA_DAILY_IS_RUNNING],
28                "errorMsg" => BatchErrorCode::AGODA_DAILY_IS_RUNNING->getMessage(),
29            ];
30        }
31
32        $this->logger->info('$existingRecord: ' . json_encode($existingRunningRecord));
33
34        /*
35         * [B]
36         * FEED_ID 32 API는 페이지네이션 방식으로 데이터를 조회한다.
37         * 따라서 다음의 초기 값을 가지고 순회를 진행하다 currentPage가 totalPage에 도달한 경우 다음 페이지가 없는 것으로 간주하고 루프를 종료한다.
38         * - currentPage = 1
39         * - total = 2
40         *
41         * FEED_ID 32 응답 데이터 中 page 데이터 형식 예시
42         * "page": {
43         *      "id": 1,
44         *      "total": 1
45         * }
46         */
47        $tobeUpdatedHotelIdsArray = []; // 호텔 IDs 누적 저장
48        $record = null;
49        while ($this->currentPage <= $this->totalPage) {
50            // Content API(FEED_ID: 32) 요청
51            $record = $this->updatedHotelIdsReader($mtypeid, $mdate, $tobeUpdatedHotelIdsArray);
52
53            $tobeUpdatedHotelIdsArray = $record->hotelIds;
54        }
55
56        if (!isset($record)) {
57            $message = "[SUCCESS] 업데이트가 필요한 호텔 ID 목록이 없습니다. 배치를 정상적으로 종료합니다.";
58            $this->logger->info($message);
59            return [
60                "result" => true,
61                "errorMsg" => $message,
62            ];
63        }
64
65        // DB.product_hotel_supplier.ota_daily_batch_execution_status 테이블에 적재
66        $saveRecordId = $this->insert($record);
67        $this->logger->info("[SUCCESS] 업데이트가 필요한 호텔 ID 목록 적재 배치 수행을 마쳤습니다. 배치를 정상적으로 종료합니다. 파라미터(mtypeid: {$mtypeid}, mdate: {$mdate})");
68
69        return [
70            "result" => true,
71            "data" => ["saveRecordId" => $saveRecordId]
72        ];
73    } catch (\Exception $e) {
74        $this->logger->info("[FAILED] 에러가 발생했습니다. 에러 메시지: " . $e->getMessage());
75        throw $e;
76    }
77}

[A]: 중복 실행은 Airflow DAG를 실행하는 인스턴스에서 막긴 하지만, 또 다른 인스턴스가 띄워지고 실행되는 혹시 모를 불상사를 막기 위해 진행 중인 배치 프로세스가 없는 지 체크하고, 있는 경우 조기 종료합니다.

[B]: 업데이트가 필요한 호텔 ID 목록이 몇 천 개가 넘을 수가 있습니다. 따라서 OTA는 페이지네이션 형태로 데이터 제공합니다. id가 total 페이지 숫자보다 작은 경우 while 루프로 순회하며 id 값을 증가시키며 API를 호출해 모든 페이지 내용을 조회하고 저장하도록 합니다.

호텔 상세 정보 업데이트 배치

다음은 적재한 업데이트가 필요한 호텔 ID 목록을 대상으로 각 호텔의 전체 정보를 조회해 Supplier 라고 하는 중간 DB에 저장하는 절차의 코드 중 일부입니다. 이는 Origin 이라고 하는 최종 DB에 마이그레이션 하기 이전의 저장소입니다.

1/**
2 * FEED_ID(32) Load 작업 마친 후 업데이트가 필요한 호텔 ID 목록 기반 호텔 상세 정보 업데이트 Job
3 *
4 * ** Agoda Affiliate API 단건 조회, 단건 적재 & 단건 마이그레이션 **
5 * 1. [Extract] hotelIds 를 기준으로 순회하며 Feed 19, Feed 10, Feed 31, Feed 14 조회
6 * 2. [Transform & Load] hotel, room, facility 마이그레이션 수행
7 */
8public function updateTobeUpdatedHotelsJob(int $tobeUpdateRowId, LanguageCode $languageCode)
9{
10    try {
11        /*
12         * 1.  ota_daily_batch_execution_status에서 id & status(EXTRACTED)를 기준으로 업데이트 하고자 하는 호텔 id 목록 조회
13         * 2.  추출된 hotelIds 목록을 기반으로 맨 앞 요소부터 업데이트 수행
14         * 3.  (FEED_ID 19) Hotel Full Info  조회 & 적재
15         * 4.  (FEED_ID 10) Hotel Other Info 조회 & 적재
16         * 5.  (FEED_ID 31) Hotel Local Info 조회 & 적재
17         * 6.  (FEED_ID 14) Hotel Facility Info per roomtype 조회 & 적재
18         * 7.  Origin -> Supplier Master   마이그레이션
19         * 8.  Origin -> Supplier Facility 마이그레이션
20         * 9.  Origin -> Supplier Room     마이그레이션
21         */
22
23        // 마지막 배치 수행 Job name
24        $lastJobStep = AgodaLastJobStep::NONE;
25        // 업데이트 대상 OTA 타입
26        $updateTargetOtaType = OtaType::AGODA;
27
28        // DB에 업데이트 완료된 호텧 IDs 저장 용도
29        $updatedHotelIds = [];
30
31        /** [1] */
32        $tobeUpdatedHotelsArray = $this->findTobeUpdatedByIdAndStatus($tobeUpdateRowId, BatchProcessType::EXTRACTED);
33        if (empty($tobeUpdatedHotelsArray)) {
34            $msg = "[FAIL] 해당 ota_daily_batch_execution_status recordId: {$tobeUpdateRowId}, status: EXTRACTED 와 일치하는 레코드가 존재하지 않습니다. 배치 프로세스를 종료합니다. LOAD 절차를 수행하기에 앞서 배치 상태는 반드시 EXTRACTED 상태여야 합니다.";
35            $this->logger->info($msg);
36            return [
37                "result" => false,
38                "errorMsg" => $msg
39            ];
40        }
41        // hotelIds string -> array
42        $tobeUpdatedHotelsArray['hotelIds'] = json_decode($tobeUpdatedHotelsArray['hotelIds'] ?? [], true);
43
44        /** [2] */
45        while (true) {
46            // 맨 앞 요소 추출
47            $hotelId = array_shift($tobeUpdatedHotelsArray['hotelIds']);
48
49            // 배치 종료 도달점 체크
50            if (empty($hotelId)) {
51                $message = "[SUCCESS] 업데이트 처리가 필요한 Agoda 호텔 및 객실 업데이트 배치 프로세스를 완료했습니다. 배치 프로세스를 종료합니다. 저장한 agoda_tobe_update_hotel id($tobeUpdateRowId), language({$languageCode->label()}), mtypeid({$tobeUpdatedHotelsArray['mtypeid']}), mdate({$tobeUpdatedHotelsArray['mdate']})";
52                $this->logger->info($message);
53
54                // ota_daily_batch_execution_status (status -> SUCCESS), 마지막 배치 Job Step, 업데이트된 호텔 IDs 업데이트
55                $this->updateTobeUpdatedById(
56                    $this->convertObjectToArray(
57                        BatchSupplierAgodaTobeUpdateHotel::ofSuccess(
58                            $tobeUpdateRowId,
59                            $updatedHotelIds,
60                            BatchProcessType::SUCCESS,
61                            $lastJobStep)));
62
63                return [
64                    "result" => true,
65                    "message" => $message
66                ];
67            }
68
69            // 기존 레코드 조회
70            $updateTargetHotelRecordId = $this->findIdAndHotelIdAndCountryIdByHotelId($hotelId, $languageCode)['id'] ?? null;
71
72            // 가져온 호텔 id 목록을 기준으로 Agoda Affiliate FEED_ID 19 호출
73            // 업데이트 수행 전 ota_daily_batch_execution_status (lastJobStep -> [1. SUPP_FULL_INFO]) 로 업데이트
74            $lastJobStep = AgodaLastJobStep::ORIGIN_FULL_INFO;
75
76            /** [3] FEED_ID(19) Hotel Full Info */
77            $hotelFullInfoJsonStr = $this->callContentAPI(ContentAPIRequest::toHotelInfoQueryParams($this->FEED_ID_HOTEL_FULL_INFO, $languageCode, $hotelId));
78            $hotelFullInfo = BatchAgodaHotel::fromHotelFullInformationJson($hotelFullInfoJsonStr);
79            $this->upsert($this->TABLE_PRODUCT_HOTEL_SUPPLIER_AGODA_HOTEL . $languageCode->label() . '.hotel', $hotelFullInfo ? [$this->convertObjectToArray($hotelFullInfo)] : []);
80
81            /** [4] FEED_ID(10) Hotel Other Info */
82            $lastJobStep = AgodaLastJobStep::ORIGIN_OTHER_INFO;
83            $hotelOtherInfo = BatchAgodaHotel::fromHotelOtherInformationJsonForUpdate(
84                $this->callContentAPI(ContentAPIRequest::toHotelInfoQueryParams($this->FEED_ID_HOTEL_OTHER_INFO, $languageCode, $hotelId)),
85                $updateTargetHotelRecordId);
86            $this->upsert($this->TABLE_PRODUCT_HOTEL_SUPPLIER_AGODA_HOTEL . $languageCode->label() . '.hotel', $hotelOtherInfo ? [$this->convertObjectToArray($hotelOtherInfo)] : []);
87
88            ... 생략 ...
89        }
90    } catch (\Exception $e) {
91        // 배치 실패 지점(hotelId), lastBatchStep 로깅
92        $message = "[FAIL] $lastJobStep->value 단계에서 예외가 발생했습니다. hotelId($hotelId), ERROR MESSAGE: {$e->getMessage()}. 배치 프로세스를 종료합니다.";
93        $this->logger->error($message);
94
95        // ota_daily_batch_execution_status status를 FAILED 로 변경
96        $this->updateTobeUpdatedById(
97            $this->convertObjectToArray(
98                BatchSupplierAgodaTobeUpdateHotel::ofFailure(
99                    $tobeUpdateRowId,
100                    $updatedHotelIds,
101                    $hotelId,
102                    $message,
103                    BatchProcessType::FAILED,
104                    $lastJobStep)));
105
106        return [
107            "result" => false,
108            "errorMsg" => $message
109        ];
110    }
111}

[1]: 바로 직전의 업데이트 필요 호텔 ID 목록 적재 절차를 정상적으로 마친 상태인지 확인하고, 만약 그렇지 않을 경우 배치를 조기 종료합니다. 업데이트가 필요한 호텔 ID 목록이 존재하지 않으면 호텔 데이터 업데이트 배치를 진행할 필요가 없기 때문입니다.

[2]: 본격적인 호텔 데이터 업데이트 절차를 시작합니다.

[3]: 해당 호텔의 상세 정보를 API를 호출을 통해 불러와 Agoda 호텔 객체로 만들고 테이블에 upsert 합니다.

[4]를 포함한 그 이후 절차: 기타, 지역, 시설 정보 등의 데이터를 원장 테이블에 업데이트하고, 중간 테이블 Supplier 로 마이그레이션 합니다.

예외 케이스에 대한 대응

Agoda 뿐만 아니라 모든 OTA에 대한 배치 프로세스는 실행하는데 오랜 시간이 걸리므로 실행 도중 문제는 없었는지, 업데이트 성공 혹은 실패한 호텔 ID는 무엇인지, 에러가 발생했다면 에러 메시지는 무엇인지 알기 위해 최대한 중요하다고 생각되는 요소는 모두 기록하려고 노력했습니다. 사람의 직접적인 개입 없이 마치 한 명의 사람처럼 스스로 생각하고 대응할 수 있어야 하는 것이 이상적인 배치라고 생각하기에, 예외 케이스를 최대한 생각해 내고 이에 적절히 대응할 수 있는 Fail Over 로직을 작성하기 위해 끊임없이 고민하고 로직에 녹여냈습니다.

호텔 도메인에 대한 이해

마지막으로 배운 점으로 호텔 시스템의 생태계에 대해 어느 정도의 이해도가 생긴 것입니다. 프로젝트 막바지에 들어서면서 늘 보던 데이터지만 그 데이터가 어떤 의미를 갖는지, 어떻게 활용될 수 있는지 알게 되었습니다. 알루랩이라는 스타트업에 들어오게 된 계기가 호텔 시스템을 개발하기 위함이였기에 개발을 진행하면서 이러한 호텔 시스템 전반에 두루두루 이해가 생겨 흥미롭고 재미있었습니다.

Tech Stack

OTA 자동화 파이프라인과 레거시 어드민을 현대화하기 위해 PHP 워커, Airflow DAG, Redis/OpenSearch 조합을 적용했습니다.

PIPELINE

Data Pipeline & Batch Automation

Airflow DAG와 PHP 워커를 결합해 OTA 데이터를 병렬로 수집, 정제하고 Failover 지점을 관리했습니다.

Apache AirflowPHP Batch WorkersRedisMariaDBOpenSearch

BACKOFFICE

Admin Experience & Domain Modeling

정규화된 호텔/객실 모델과 어드민 UI 흐름을 정의해 신규 데이터 생성과 검수 속도를 높였습니다.

Domain ModelingSmarty TemplatesLegacy UI IntegrationDesign System Adoption

OPS

Operations & Monitoring

텔레그램 알림, 운영 대시보드, 배치 모니터링을 통해 현지 운영팀이 즉시 대응하도록 가시성을 마련했습니다.

Telegram BotOperational DashboardsAlerting WorkflowRollback Playbooks