스트림 데이터를 가공해 원하는 형태로 변환: map과 scan으로 만드는 데이터 파이프라인

1. 개념

변환 연산자(transformation operator)는 Observable이 방출하는 값을 받아서 다른 값이나 다른 구조로 바꿔주는 함수다. 대표적으로 map은 각 값을 새로운 값으로 매핑(mapping)하고, scan은 이전 결과와 현재 값을 합쳐 누적된 상태를 만든다. 여기에 filter 같은 필터링 연산자를 섞으면 조건에 맞는 값만 골라 가공할 수 있다.

이 실습에서는 원본 데이터 스트림 하나를 받아서 map → filter → scan 순서로 파이프(pipe)를 구성해, 최종적으로 화면이나 다음 로직에서 바로 쓸 수 있는 형태로 변환하는 과정을 만든다.

2. 왜 사용하는가

Observable이 방출하는 원본 데이터는 대부분 그대로 쓰기 어렵다. 서버 응답은 필드명이 다르거나 불필요한 값을 포함하고, 사용자 이벤트는 좌표나 타임스탬프 같은 부가 정보가 섞여 있다. 매번 구독(subscribe) 콜백 안에서 이런 가공 로직을 처리하면 코드가 길어지고 재사용이 어렵다.

변환 연산자를 파이프라인 형태로 구성하면 "데이터가 어떻게 흘러가면서 바뀌는지"를 선언적으로 표현할 수 있다. 구독 콜백은 이미 가공된 최종 데이터만 받아서 처리하면 되므로 관심사가 분리된다.

3. 동작 원리

map(fn)은 소스 Observable이 값을 방출할 때마다 fn을 호출하고, 그 반환값을 새로운 Observable로 다시 방출한다. 값 하나가 들어오면 값 하나가 나가는 1:1 구조다.

filter(predicate)는 predicate가 true를 반환하는 값만 다음 단계로 흘려보낸다. 조건에 맞지 않는 값은 그 자리에서 사라지므로 이후 연산자는 필터링된 값만 받는다.

scan(accumulator, seed)은 내부에 누적 상태를 하나 들고 있다. 값이 방출될 때마다 accumulator(누적값, 현재값)을 호출해서 새로운 누적값을 만들고, 그 결과를 즉시 다음 단계로 방출한다. reduce와 계산 방식은 같지만, reduce는 스트림이 끝날 때 한 번만 결과를 내보내고 scan은 매 값마다 중간 결과를 내보낸다는 차이가 있다.

이 세 연산자를 pipe 안에 순서대로 배치하면, 값이 위에서 아래로 흐르면서 매핑, 필터링, 누적이 차례로 적용된다.

4. 예제

먼저 const { from } = require('rxjs');와 const { map, filter, scan } = require('rxjs/operators');로 필요한 모듈을 불러온다.

그다음 원본 데이터를 준비한다. const orders = [{id: 1, price: 12000, status: 'paid'}, {id: 2, price: 8000, status: 'canceled'}, {id: 3, price: 30000, status: 'paid'}, {id: 4, price: 5000, status: 'paid'}]; 형태로 주문 목록을 배열로 만든다.

이제 from(orders)로 배열을 Observable로 바꾼다. from은 배열의 각 요소를 순서대로 하나씩 방출하는 Observable을 만들어준다.

여기에 .pipe(...)를 연결한다. 첫 번째 단계는 filter(order => order.status === 'paid')로 결제 완료 상태인 주문만 남긴다.

두 번째 단계는 map(order => order.price)로 각 주문 객체에서 가격 값만 추출한다. 이 시점부터 스트림에는 순수한 숫자 값만 흐른다.

세 번째 단계는 scan((total, price) => total + price, 0)으로 지금까지 누적된 결제 금액 총합을 계산한다. 0은 누적을 시작하는 초기값(seed)이다.

전체를 이어붙이면 from(orders).pipe(filter(order => order.status === 'paid'), map(order => order.price), scan((total, price) => total + price, 0)).subscribe(total => console.log('현재까지 누적 금액:', total)); 형태가 된다.

이 코드를 실행하면 결제 완료 주문이 들어올 때마다 누적 합계가 순서대로 출력된다. 첫 번째 결제 완료 주문에서는 12000이 출력되고, 두 번째 결제 완료 주문에서는 42000, 세 번째에서는 47000이 출력된다. 취소된 주문은 filter 단계에서 이미 걸러졌기 때문에 계산에 포함되지 않는다.

5. 실무 사용 사례

REST API 응답을 화면 컴포넌트가 요구하는 형태로 바꿀 때 map을 쓴다. 서버가 {user_name, user_age} 같은 snake_case 필드를 내려줘도 map(res => ({name: res.user_name, age: res.user_age}))로 즉시 정규화할 수 있다.

실시간 로그나 센서 데이터를 처리할 때 scan으로 이동 평균, 누적 카운트, 최댓값 갱신 같은 상태를 유지한다. 별도의 외부 변수를 두지 않고 스트림 내부에서 상태를 관리할 수 있어 구독 콜백이 단순해진다.

폼 입력이나 검색 이벤트 스트림에서는 filter로 빈 문자열이나 유효하지 않은 입력을 먼저 제거하고, 그 뒤에 map으로 요청에 필요한 파라미터 형태로 변환하는 조합이 자주 쓰인다.

6. 주의할 점

map과 scan에 넘기는 콜백 함수 안에서 외부 변수를 직접 변경하는 부수 효과(side effect)를 넣지 않는다. 변환 로직은 입력값만으로 출력값이 결정되는 순수 함수로 유지해야 디버깅과 테스트가 쉬워진다.

scan의 초기값(seed)을 빠뜨리면 첫 번째 방출값이 그대로 누적값으로 취급되어 의도한 계산 결과와 달라질 수 있다. 합계나 카운트처럼 명확한 초기 상태가 필요한 경우 반드시 seed를 지정한다.

원본 객체를 직접 수정하지 않는다. map(order => { order.price = order.price * 1.1; return order; })처럼 원본 객체의 속성을 바꾸면 다른 구독자나 뒤쪽 파이프라인이 예상치 못한 값을 보게 될 수 있다. 새로운 객체를 만들어 반환하는 방식(map(order => ({...order, price: order.price * 1.1})))이 안전하다.

7. 핵심 정리

변환 연산자는 스트림 안을 흐르는 값을 원하는 형태로 바꾸는 도구다. map은 값을 다른 값으로 매핑하고, filter는 조건에 맞는 값만 남기고, scan은 값들을 누적해 상태를 만든다. 이 세 연산자를 pipe 안에서 순서대로 조합하면 원본 데이터가 최종 형태로 변환되는 과정을 선언적인 코드로 표현할 수 있다. 콜백 함수는 순수 함수로 작성하고 원본 데이터를 직접 수정하지 않는 것이 안전한 사용의 기본 원칙이다.