[데이터 엔지니어링 디자인 패턴] 05. 데이터 가치

데이터 엔지니어링 디자인 패턴 시리즈의 글입니다.

스토리지 어딘가에 저장된 데이터는 진짜 자산이 아니다. → 결국 데이터 품질에 문제가 생김..
데이터셋을 강화하여 최종 사용자에게 더 쓸모있게 만드는 작업이 필요하다.

5.1 데이터 강화

원시데이터는 기술적 제약으로 품질이 떨어짐
ex. 이벤트 = 확장된 컨텍스트가 부족
Data Enrichment 패턴은 이러한 제한을 극복
데이터를 다양한 이해관계자에게 더 유용하게만듦

정적 조이너

Static Joiner Pattern 강화데이터셋의 본질이 정적이라면 이 방법을 선택

문제
- 사용자 활동데이터에 사용자 참조데이터를 조인해서 서비스하고싶음.
해결책
- 데이터셋을 결합하는데 사용할 속성 목록을 선택 (join on 컬럼)
- 강화 데이터셋이 천천히 SCD 테이블이라면, 시간제약이 발생
  - Slowly Changing Dimensions (SCD) : 천천히 변화하는 차원
  - 시간의 경과에 따른 엔티티의 변화를 지원하는 모델링 전략
  - 시간에 따른 레코드 변화를 추적 (=즉, history테이블)
  - ex. 한 테이블에 쌓고 is_current 테이블로 최종 컬럼 표기
  - recent / history 테이블 각각 쌓고 컬럼으로 조인
- API를 활용해 데이터셋을 강화하기도 함.
  - 멱등성이 필요하다면, API 데이터를 한번 내부 스토리지에 저장해서 구체화 한뒤
  - 조인해서 사용.
결과
- 지연 데이터와 일관성
  - 때때로는 history테이블에 최신 변경사항이 포함되지 않는 ‘지연 문제’가 생김
  - 스트리밍 파이프라인이라면? → Dynamic Joiner pattern 사용
  - 배치 파이프라인이라면? → ochestration을 통해 강화 데이터셋을 기다림.
- 멱등성

동적 조이너

Dynamic Joiner pattenr 동적 환경에 더 잘 적응된 대안

문제
- 각 사용자 변경사항이 CDC 패턴에서 스트리밍 브로커에 등록됨
- 이러한 스트리밍 파이프라인 상황에서 데이터를 결합할 방법이 필요함
해결책
- 키의 식별 / 조인메소드 정의 / 시간경계 선택
- 전용 시간관리 전략을 선택하지 않으면 두 데이터셋의 지연 시간이 서로 다르기 때문에 Join이 비게 될 위험이 있음
- 양쪽 스트림에서 조인된 레코드의 Buffer에 시간 경계를 설정.
- Buffer는 조인이 일어날 수 있도록 추가시간을 제공
- Buffer는 GC watermark를 포함 (= 오래된 이벤트가 버퍼에서 언제 사라져야하는지를 정의)
- GC watermark 사용 시점을 정의해야함.
- 지연시간이 더 느린 스트림B의 가장 오래된 이벤트시간보다 오래된 스트림A의 레코드는 제거.
결과
- 공간 vs 정확성 trade-off
  - 버퍼공간을 늘리면 놓치는 데이터가 줄지만, 스토리지 비용이 비싸짐
  - 버퍼공간을 늘리면 스토리지가 최적화되지만 지연이 큰 데이터는 놓치는 경우가 많아짐
  - ⇒ 두가지 요소의 균형 필요
  - ⇒ 해결책은 비즈니스 요구사항과 조인된 데이터셋의 지연 차이에 따라 달라짐
- 지연 데이터
  - 스트림 처리는 낮은 지연 처리 시맨틱으로 인해 지연데이터 통합에 대한 내결함성이 약함
  - ex. 스트림이 일부지역에서 일시적 연결문제를 겪는다면..?
  - 결국 이 제한을 극복하려면 지연 데이터를 추적하고 통합해야함

5.2 데이터 데코레이션

데이터셋이 강화 패턴으로 가치를 높였다면, 이제는 최종 형태로 만들 차례
이러한 상황이라면 Data decorator pattern이 유용하다.

래퍼

Wrapping = 객체에 별도의 동작이나 속성을 추가 레코의 원본 / 변환된 부분을 분리하는데 도움이 되는 패턴.

문제
- 스트리밍 방문데이터를 처리하고자함
- 방문데이터는 다양한 provider에서 제공되는데, 출력 스키마가 서로 다름
- 다양한 필드를 단일위치에 넣어서 downstream consumer가 쉽게 처리할수 있도록 하려함
- 처리로직의 단순화를 위해 raw data와 computed data를 명확히 분리하되 디버깅요구를 위해 원래 구조는 유지해야함
해결책
- 원본레코드를 그대로 유지해야한다면? ⇒ Wrapper Pattern 사용
- 레코드 수준에 또 다른 추상화를 추가
- raw / computed 필드로 구성된 이벤트로 변환
- Avro, Protobuf, Json 형식과 테이블같은 구조화된 형식 모두 지원
- 별도의 테이블로 구현 / 테이블 내의 추가컬럼 구현 모두 가능
- 스키마 관리 필요
결과
- 도메인 분할
  - raw / computed 모두에서 관련 필드를 찾을 수 있음
  - 데이터 조회가 복잡해짐
  - 데이터가 두 위치에 나누어져있게됨
  - 래핑데이터를 실버계층에 저장하여 최종데이터로 노출되지 않게 하면 좋음
- 크기
  - decorated값은 추가된 값이므로, 전체 데이터 크기와 네트워크 트래픽에 영향을 줌
  - 관심있는 컬럼 선택 / 해당 컬럼만 접근하는 방식으로 완화 가능

메타데이터 데코레이터

Metadata Decorator Pattern 추가된 값이 사용자에게 노출되지 않아야한다면 데이터 스토어의 메타데이터 계층에 추가레코드를 숨긴면 된다.

문제
- 스트리밍 잡이 매주 새 버전을 출시
- 생성된 각 레코드에 job version같은 기술적 context를 추가해야함
- 그러나, 최종 사용자에게 보내는 record에는 포함시키고 싶지 않음
해결책
- 데이터 스토어의 metadata 계층을 활용
- 데이터 스토어의 메타데이터 처리 역량에 따라 구현이 달라짐
- 각 레코드를 전용 메타데이터 속성과 연결
결과
- 구현
  - 스토리지들의 메타데이터 지원 부족 ex. Amazon Kinesis
  - 메타데이터 정보를 처리하기위해 추가 컬럼이나 테이블을 정의해야하는 경우가 많음
- 데이터
  - 비즈니스 관련 속성을 작성하는것은 삼가야함
  - 메타데이터는 데이터에 관한 데이터라는부분을 명심

5.3 데이터 집계

정보를 제거하는것도 데이터 가치를 창출하는 방법이 될 수 있다.

분산 집계기

Distributed Aggregator 분산 데이터 처리 프레임워크를 활용 물리적으로는 독립 / 논리적으로는 유사한 항목을 결합

문제
- 방문데이터 온라인 분석 처리 (OLAP) 큐브를 구축하여 방문 데이터를 대시보드 시나리오에 맞게 집계형식으로 변환하고자함
- 데이터셋은 일별 이벤트 시간 파티션에 저장
- 분석 큐브는 일별 및 주간 뷰를 표현
해결책
- 보통 group by 로 충분
- 단, 빅데이터는 관련 레코드가 여러 문리적 장소에 분산 저장
- ⇒ Distributed Aggregator Pattern 사용
- 여러 장비를 활용하여 Cluster (단일 실행 단위)를 형성
- 각 클러스터는 빅데이터 작업을 나눠서 분담함
- Shuffle = 분산 집계기의 실행에서 서로 다른 장비에 처음 적재된 레코드를 네트워크를 통해 교환하는 단계가 포함됨.
- 셔플은 네트워크 트래픽 비용으로 자주 지연을 일으킬 수 잇음
- 셔플 전에 로컬에서 부분집계를 수행하여 최적화 가능
- 교환 레코드들의 크기가 작아지고 전체 작업이 빨라짐
- (ex. 계수연산 - 초기노드에서 부분 카운트 수행, 마지막에 합산 )
결과
- 추가적인 네트워크 교환
  - 총 2번의 네트워크 교환 발생
    1. 입력 데이터를 각 노드에 전달
    2. 같은 서버에서 관련 데이터를 수집
- 데이터 스큐(Skew)
  - 특정 키가 다른 키들보다 발생 빈도가 훨씬 큰 불균형한 데이터셋
  - salting기법으로 스큐 방지 가능
- 확장
  - 노드는 계획된 reduce작업을 완료 해도, 내결함성 이유로 하드웨어계층에서 사용될 수 있음
  - 만약 reduce 연산이 실패하여 재시작되면 해당 노드는 계속 존재하게되며 회수되지 않음
  - ⇒ shuffle service (셔플데이터를 저장하고 서빙하는 추가 컴퓨팅 요소) 로 노드를 관리

로컬 집계기

Local aggregator 데이터셋이 입력에서 올바르게 파티션되거나 단일장비에 적합하다면 네트워크 교환은 필요 없다

문제
- 스트리밍잡은 파티셔닝된 스트리밍 브로커에 저장된 방문 window생성
- partition은 변형이나 변화가 없을예정
- 이때 job을 최적화하고 Grouping shuffle을 제거하고싶다면?
해결책
- local partitionng을 사용한다면
  - 비용이 많이 드는 shuffle제거
  - 정적 데이터 원천 partitioning
  - 관련속성의 공동 저장
- 집계를 로컬에서 수행 가능
- 단, 고정파티션 스키마와 올바른 입력데이터 분산이라는 엄격한 가정하에서 수행 가능
- 태스크가 fully isolated되어 다른 태스크를 기다릴 필요가 없음
- ⇒ 일부 느린 처리 유닛이 전체 실행을 지연시키는 단점이 없어짐
- 단, producer쪽에서 구현 노력을 들여야함. (특정 grouping key가 동일한 물리 파티션에 기록)
- consumer에 사전 파티션 데이터셋을 셔플형식으로 변환하는 함수들 제공
  - groupByKey
  - mapPartitions
  - foreachPartitions
결과
- 확장
  - 원천이 정적이어야하고, 일관된 파티션을 전제하기에 확장이 어려움
  - 확장/조정이 필요하면 모든 레코드의 파티션 할당을 새로해야함
  - stop-the-world 필요
  - 셔플을 피하지만 확장이 훨씬 어려워진다는 단점
- 그룹핑 키
  - 모든 컨슈머가 공통의 그룹핑 키 로직 사용하길 기대함
  - 컨슈머가 각자 다른 집계를 사용하게된다면?
  - ⇒ 동일 레코드를 각가 다른 그룹핑 키로 작성해야하는 상황 초래

5.4 세션화

session = 같은 활동과 관련도니 이벤트를 결합하는 특별한 종류의 집계기
시작지점 / 세션이벤트 / 종료지점으로 구성
정지상태 / 이동중 상태에 따라 패턴 선택 가능

증분 세션화 처리기

Incremental Sessionizer Pattern 배치파이프라인에서 사용됨

문제
- 방문이벤트를 매시간 파티션된 위치에 저장
- 해당 사용자가 2시간 내에 다시 방문하지않으면 세션 종료로 집계
- 한번의 방문이 최대 3개의 다른 파티션에 저장 가능
- ⇒ 데이터 분석에 어려움.
해결책
- 하나의 세션에 대한 레코드가 여러 연속된 파티션에 존재 = Incremental Sessionizer Pattern
- 세 가지 스토리지 공간 설정
  - 입력 데이터셋 스토리지 : raw event 저장
  - 완료된 세션 스토리지 : 완료된 세션 기록
  - 대기 중인 세션 스토리지 : 여러 파티션에 걸쳐있는 세션을 모두 저장
- 워크플로 정의
  - 입력데이터셋과 이전 모든 대기세션 결합
  - 세션화 로직 적용
    - 초기화 : 세션시작
    - 축적 : 세션 활성
    - 완료 : 세션 중지
  - 완료된 세션은 공개 노출 스토리지로 이동
  - 대기세션은 다음 잡 실행을 위해 계속 생존.
결과
- 비활성 기간
  - 세션을 얼마나 오래 열어둘 것인가?
  - ⇒ 적절한 균형, trade-off
  - 계속 진행되는 세션에 플래그를 표시해야함 ( ex. is_completed = False)
- 데이터 최신성
  - 배치 파이프라인안에서 작동
  - 인사이트가 실시간에비해 늦게 도출
- 지연데이터, 이벤트 시간 파티션, 백필링
  - 이미 처리된 파티션에 대한 세션을 놓칠 수 있음. → 지연데이터 문제
  - 지연데이터를 통합해도 다음 세션에 영향을 미칠 수 있음
  - 백필링시 모든 파티션에 대해 동일작업을 해야함

상태 저장 세션화 처리기

Statefule Sessionizer 데이터 최신성이 문제면 더 빈번하고 작은 반복을 하는 세션화 패턴을 사용해야함

문제
- 더 낮은 지연으로 세션에 접근할 필요가 생김
- 스트리밍 브로커에서 몇 초내에 방문정보 확인이 가능
- 세션을 거의 실시간으로 생성하고싶음
해결책
- 스트리밍 파이프라인도 상태 비저장임.
- Statuefule Sessionizer Pattern은 state store(상태 스토어) 라는 추가컴포넌트를 활용
- 대기세
결과