개발 갈무리

1. 서론

 - 재화나 서비스를 판매하는 기업은 판매 채널을 가지고 있다. 자사채널, 메타서치, 메타부킹 등으로 나뉘지만 본질적으로 마케팅 관점에서는 display, click이 일어나는 공간이고, customer의 입장에서는 view, click, book(purchase)를 하는 곳이다.

 - IT에서 대부분 이런 행위는 API나 EDI 등으로 연결된 Data Pipeline에 의해 일어난다. 결국 Request - Response 구조다.

 - 문제는 Request - Response / View - Click - Book 사이에 일어나는 데이터 액션이 1:N 또는 1:N:N 같은 형태로 일어난다는 점이다.

 - 심지어 각 scope에서도 semantic normalization → label unification → aggregation 과정이 필요하다. (결과적으로 Response는 같은데 Request 조건이 다름. 가령 항공권을 검색할 때, ICN-NRT나 SEL-NRT는 같은 Response)

2. 본론

a. 1:1로 정보를 degrade 시키는 경우

 - 1:N으로 일어난다고 하더라도 Business 요구사항에서는 직관적인 1:1을 선호한다. 결국 big data로는 큰 흐름을 보고자 하기 때문이다. 직관적이라 의사결정자에게 어필하기 쉽다는 것도 큰 장점이다.

 - 1:1로 정보를 degrade 시킬 때 가장 좋은 방법은 request에 있는 요청 정보들 중 키값을 조합해 대표성을 띠는 유니크 키로 만드는 것이다. 가령 채널 + 검색어 + 검색(필터)조건 등이다.

 - 1:1로 만들기 때문에 다른 어떤 케이스보다 semantic normalization → label unification → aggregation가 중요하다. 특히 TOP N 같은 metric을 본다면 필수다.

b. 1:N으로 붙이는 경우

 - 1:N으로 붙이면 raw data를 그대로 밀어넣기 때문에 ETL 과정은 편할 수 있다. ETL 관점에서는 capacity가 거의 유일한 관리 point이다.

 - 데이터 사용자 기준에서는 관계를 1:N으로 만드는 dimension을 꼭 모두 넣어서 분석해야 한다. 그렇지 않으면 request의 metric들이 중복으로 계산되기 때문이다.

 - 다만 중복으로 계산하더라도 정확한 지표가 아닌 Volume index로 보고 상대적 크기 지표로 활용한다면 괜찮을 수도 있다. 가령 N에 해당하는 것도 넓게 생각하면 display 되는 scope으로서 가중치를 부여할 수 있기 때문이다. (더 포괄적인 검색어에 대해 더 높은 가중치 부여해 business 의사 결정)

 - 특히 데이터가 복잡한 경우 고의로 request to book(purchase)을 만드는 게 더 직관적일 수 있다. 그래도 분석 기간은 단기로 잡는 게 좋다. 보통은 시장의 시즌이 존재하기 때문이다.

 - 1:N:N도 필요하다면 확대해서 쪼개고 컨셉마다 dataset을 만드는 것이 좋을 수 있다. 위에 설명한 것 같이 기간은 되도록 너무 길지 않게, dimension 생략에 따른 부작용을 잘 이해해야 한다.

c. N:N으로 붙여야 하는 경우

 - 아무리봐도 N:N으로 느껴진다면 semantic normalization → label unification → aggregation 과정을 거치지 않았거나, 응답간의 교집합이 자주 존재하는 경우다. 최근에는 AI를 통해 semantic normalization이 가능한 영역이 더 넓어졌다고는 하지만, 분명히 다르게 봐야 되는 케이스들이 존재한다.

 - 가령 검색에서 제주 놀거리, 제주 여행, 제주 투어, 제주 액티비티를 검색한다고 하면, 어떤 것들을 같은 것으로 묶고 어떤 것은 다른 것으로 묶을지 쉽게 판단하기 어렵다. request의 intent를 통일시킬 수 있는 정보가 부족하기 때문이다.

 - 그래서 이런 데이터는 큰 topic 단위로 세분화시켜 다양한 dataset의 컨셉을 만들어야 한다. 사업이 상대적으로 영세하고 검색데이터가 크지 않다면, 이런 분석은 후순위로 밀어두는 것이 더 낫다. 반대로 사업 규모가 크다면 기획실로부터 다양한 컨셉들이 나와 각 사업부서의 대시보드가 꾸려져야 한다.

 - 대부분은 좀 더 실무자에게 적합한 대시보드일 가능성이 높다. 위로 report 된다면, 보통은 trend 추적을 위해서이며  영역별 TOP N개가 선택된다.

3. 결론

 - 거의 대부분의 마케팅 조직, 기획 조직에서 활용하는 데이터기 때문에, ETL을 주로 하는 포지션일지라도 도메인에서 무엇을 쪼개어 분석하고자 하는가에 관심을 가져야 이런 설계를 진행할 수 있다.

 - 특히 이 고민은 최소한 datalake에서 BI로 넘어가는 상황에서 필수적으로 해야 한다. 이 컨셉을 각각 따로 잡고 설계하고 데이터 사용자에게 설명해주지 않으면, 도메인 유저들은 데이터를 불신하기 시작한다. 데이터 불신이 장기화 되면 데이터 없는 의사결정 습관이 생기게 되고, 직관으로 의사결정하는 것을 '경험'이라고 포장하는 문화가 만들어 진다. 데이터를 근거로 들지 않는 '경험'은 그 규모가 커질수록 위험하다. 그것이 장기화 되면 조직을 무너뜨리는 레거시가 된다.

Multi-Dimension 데이터 결합 시 자주 마주치는 문제 상황

완전히 같은 dimension으로 aggregation된 data를 결합할 때는 정합성 확보가 쉽다.

단순히 left join이나 outer join을 수행해도 무방하다.

그러나 실무에서는 어떤 dataset에 다른 dimension이 요구되는 상황이 흔하다.

예를 들어, 어떤 상품의 조회수와 구매수에 대한 dataset을 만들려고 한다.

그럼 조회수 datalake와 구매수가 있는 data warehouse에서 각각 data를 extract하고 결합하는 프로세스를 거친다.

이때 조회수를 aggregate하는 dimension과 data warehouse를 aggregation 하는 dimension을 다를 수 있다.

흔히 이런 경우에 자주 요청되는 것은 '공급업체' 같은 것들이다. '조회'를 할 때는 대부분 공급업체를 따로 요청하지 않는다.

보통은 조회 결과가 나온 후 필터되는 형태다. (물론 뭐 검색 데이터의 request에 공급업체가 필수라거나 하면 괜찮겠지만)

그렇게 되면 조회수가 공급업체 차원과 조인되면서 조회수가 공급업체 수만큼 중복 집계되는 multiply 현상이 발생한다.

이런 dataset의 사용자는 분석 시, 공급업체를 반드시 dimension으로 포함하거나, 필터링 equal 조건으로 지정해야 정확한 집계 결과를 얻을 수 있다.

결론

이렇게 결합된 데이터셋은 주로 마케팅, 사업 기획 부서에서 BI 툴을 통해 분석에 활용된다.

전환율(conversion rate) 분석과 같이 여러 dimension을 결합하는 데이터셋은 이러한 dimension 불일치 문제를 반드시 고려하도록 BI툴 사용자에게 가이드해야 한다.

상황: 특정 condition에서 보정을 위해 grouping & apply 적용

이슈: 재계산되지 않은 컬럼은 str, 재계산된 컬럼은 float인 상태에서 .str.replace 후 astype(float)으로 형변환하니 기존 float값이 모두 NaN이 됨(정확히는 기존 컬럼에서 %를 없애고 숫자로 convert하는 과정. 재계산하는 컬럼은 이미 계산을 위해 %가 없어진 상태.)

원인: accessor가 'str로 변환한 다음에 replace하고 형변환'처럼 보이지만, 실제로는 해당 시리즈가 str임을 전제하고 벡터작업을 함. 따라서 str이 아닌 값들은 type 정보를 잃어 버리게 되고 type 정보를 잃어버린 value들은 형변환 시 na값이 됨

결론: 시리즈 타입을 확신할 수 없다면 accessor 쓰기 전에 웬만하면 형변환을 명시적으로 해주도록 하자. 웬만하면 수집, 추출할 때 data source에서 type이 정렬되어 들어오긴 하지만, 그렇지 않은 data source도 있을 수 있다.

1. baseDir 및 하위 filter는 files로 대체하자

 - baseDir은 웬만하면 쓰지 말자. Directory로 지정해서 쓰면 디버깅할 때, 어떤 작업에서 정확히 어떤 parquet가 쓰였는가를 찾기가 거의 불가능하다.

 - 이것때문에 아예 baseDir에 있는 파일을 미리 삭제하거나 이동시키고 작업하는 케이스도 있는데 그럼 이제 아예 작업했던 데이터를 까볼 방법이 없어진다.

 - 특히 local에 떨군 parquet를 사용해 load한다면, directory에 있는 모든 파일을 업로드해 버린다. files를 같이 써도 둘 다 적용돼 버린다. ingestion 성능을 위해서라도 필요하다.

 - files로 설정하고 spec 로깅을 남기면 spec 로깅만 보고 바로 어떤 파일이 문제였나 알 수 있다.

 - 공식 문서 기준으로 그냥 local이 아니라 s3 같은 곳에서 가져오는 경우 files가 우선이라고 하던데, 그렇다고 한들 굳이 baseDir을 쓰는 장점이 없다. spec 작성을 손으로 할 거면 모를까.

2. transformSpec을 사용하기 보다는 전처리하고 그대로 올리자

 - data type을 위해서 dimensionsSpec을 쓰는 건 ingestion에 크게 지장이 없는데, transformSpec은 데이터가 크면 ingestion에 영향을 끼친다

 - 결과적으로 transformSpec을 쓰면 worker가 해야 할 일을 druid한테 시키는 행위가 된다. 개별 worker 단위에서 어차피 데이터를 extract해서 메모리에 load하고 있는 시점이 있을텐데 굳이 druid에 넘기는 건 합리적이지 않다. (심지어 둘 중에 일반적으로는 druid 리소스가 더 귀하지 않나 싶다.)

 - 문제는 이를 활용해 만들어진 dataset은 안 쓰고 싶다고 일부만 바꾸지는 못 하는 것처럼 보인다. 최소한 나는 그랬는데, Success로 뜨고 기존 __time에 해당하는 segment가 일부 지워지는 현상을 겪었다.

3. granularitySpec

 - segmentGranularity, queryGranularity 이렇게 2개가 핵심.

 - segmentGranularity는 데이터를 저장할 때 쪼개는 시간 단위인데 기본적으로는 data의 가장 작은 단위랑 맞춰주는 게 안전하다. 데이터 활용에 대한 요구사항을 어느 정도 숙지하고 있다면, 그 부분에 맞춰서 지정해주면 좋다. 일반적으로는 HOUR 미만으로 들어가려면 최신 데이터를 빠르게 따라가는 topic이라는 거라 파이프라인 전체 관리를 타이트하게 해야 한다.

 - Load하는 관점에서 보면, segmentGranularity가 MONTH라면 5월 레코드를 하나만 넣어도, 기존 5월 data는 모두 삭제되고 지금 넣은 레코드로 업데이트 된다. MONTH로 해놨는데 5월 30일 데이터가 이상하다? 그럼 5월 전체를 다시 넣어야 된다. 하지만 데이터 활용을 거의 주로 월단위로 한다면? 그땐 선택이다. 가용 리소스 대비 데이터가 너무 크다? 그럼 불편해도 MONTH 쓰자. 아니라면 웬만해선 DAY를 초과하지 않는 걸 추천한다.

 - queryGranularity는 rollup하는 시간 단위다. druid가 group by 성능이 좋은 가장 큰 이유는, 미리 agg를 다 해놓는다는 거다. 일종의 답안지를 미리 작성해 놓는 건데, 이건 웬만하면 raw data의 최소치에 맞추는 게 좋다. 원본이 HOUR면 HOUR로 DAY면 DAY로 하자. 집계를 해버리는 거라 데이터의 사용 입장에서 시간 해상도를 더 낮추기 어렵기 때문이다. 이건 너무 크게 하면 마치 라면을 짜게 끓인 거랑 비슷하다. 밍밍하면 소금 더 타면 되지만 짤 때 물을 더 타면 이미 그건 우리가 아는 라면이 아닐 가능성이 높기 때문.

4. "dropExisting": true

 - 이건 드루이드에 있는 약간의 변태성 때문에 효율이 생기는 옵션이다. 드루이드는 seg를 교체하면 그 전에 이 seg 뭐였어 라는 걸 다 남겨두기 때문에 이 옵션을 주지 않으면 druid 저장 공간이 빠르게 차는 경험을 하게 된다.

 - superset 같은 BI툴을 연결해 사용한다면, 더더욱 latest data만 활용하기 때문에 웬만하면 기본 옵션처럼 넣는 것을 추천한다.

 - 대부분 데이터 크기가 엄청 크지 않다면 드라마틱한 효과는 없겠으나, 드루이드 관리자가 슬퍼지지 않게 설정해주자

1. 이슈: config로 쓰고 있던 google sheet 문서가 손상되어 접근이 불가능한 상태. 사본 만들기 및 import range 함수를 통한 불러오기도 내부 에러로 실패. 중요한 hourly dag였기 때문에 바로 fix하고 배치가 돌아야 하는 상황.

2. 추정 원인: 구글 내부 에러(잦은 api 호출 등으로 막히는 상황도 고려했으나 그 정도의 대량 호출은 없었음. 구글 드라이브 인스턴스가 죽었다고 하기에는 소유자의 다른 문서들은 잘 열리는 상황. 저장중 모종의 이유로 인해 손상된 것으로 보임.)

3. 해결: daily로 google sheet로 된 모든 config를 S3로 xlsx로 백업하는 dag가 있어 해당 파일을 통해 복원. 권한 등의 적절한 복원을 위해 '편집자'는 confluence 등에 별도 작성할 필요가 있음

교훈: 외부화된 config는 손상되거나 접근이 불가능해질 경우 대처가 안 되므로 항상 백업을 진행하는 것이 좋음

cf) google sheet에 대한 api 접근은 은근히 많이 실패하는 편이다. queue limit과도 별 상관없고 짧은 기간 동안 특정 region에 할당된 인스턴스가 죽으면서 발생하는 것으로 보였다. N mins 단위로 도는 배치에서는 이중삼중으로 retry를 할 필요가 있다. 현재 진행하는 프로젝트에서는 여러 gcp project의 service account를 통해 retry를 할 수 있게 구성했다. 그럼에도 불구하고 문서 자체가 손상되면 어쩔 수가 없다. 원본 문서는 다음날 복구 되었으나, deprecated 처리했다.

현업과 소통하다 보면,

오케스트레이터에 직접 현업용 아이디를 만들어주고 권한을 컨트롤 하는 것보다,

구글시트에 config를 만드고 입력하라고 하는 게 커뮤니케이션상 훨씬 편할 때가 많다.

크게 느끼는 장점은 4가지이다.

1. 오케에서 Asset은 잘못 변경 시 audit로그를 뒤져야 하지만, 구글시트에서는 history가 바로 보인다

2. 새로운 인원이 추가되더라도 해당 인원의 아이디만 권한 허용해주면 끝이다

3. 현업들은 대부분 엑셀에 훨씬 익숙하다

4. 다른 과제의 Asset을 변경하는 등의 변수를 신경쓰지 않아도 된다. 물론 오케에서도 할 수는 있지만 시트가 훨씬 직관적이다

Asset에 해당하는 값을 여러 개 입력하기 때문에 컬럼을 그만큼 늘리는 건 가독성이 안 좋고,

Asset명 / 설정값으로 세팅하는데 그런 경우 가져오는 패턴을 기록해 본다.

dicConfig = dtConfig.AsEnumerable().ToDictionary(Function(row) row.Field(Of String)(" Asset명"), Function(row) row.Field(Of object)("설정값"))

+ 기존 config에 추가되게 하고 싶다면 For each row를 사용해 InitAllSettings에 있는 패턴처럼 하면 된다.

<for each row> dicConfig(row(" Asset명").ToString) = row(" 설정값").ToString </for each row>

collection간 비교가 필요할 때 자주 사용하는 패턴이 보여서 정리하려고 한다

1. list vs list

  i) 공통된 원소 비교 set(a) & set(b)
  ii) 완전히 동일한 리스트인지 a == b
  iii) 원소 같고 중복도 같은지 (순서 X) Counter(a) == Counter(b)
  iv) 순서/중복 무시하고 포함 여부 확인 set(a).issubset(set(b))
  v) 교집합 외 개수 비교 len(set(a) & set(b))

2. set vs set (보통은 그룹별 키가 얼마나 겹치나 볼 때 썼음)

  i) 공통된 원소 a & b, a.intersection(b) 교집합
  ii) 두 set의 모든 원소 (중복 제거) a | b, a.union(b) 합집합
  iii) a에만 있는 원소 a - b, a.difference(b) 차집합
  iv) 서로 다른 원소만 a ^ b, a.symmetric_difference(b) 대칭 차집합
  v) a가 b의 (진)부분집합인지 a <= b, a.issubset(b) 부분집합 관계 확인
  vi) a가 b의 진부분집합인지 a < b 진부분집합
  vii) 겹치는 원소가 있는지 not a.isdisjoint(b) 공통 원소 존재 여부 판단

3. df vs df

| 완전 동일 여부     | `df1.equals(df2)`               | 정렬까지 같아야 함 |
| 값 위치별 비교     | `df1 == df2`                    | 마스크 형태 결과. matrix 비교라 같은 행,렬 위치에 있는 값들끼리 == 비교한 것  |
| 행 단위 차이 추출   | `merge(..., indicator=True)`    | 유용|
| key 기준 변경 추적 | `merge` + suffix + 비교           | 특정 컬럼 비교   |
| 열 값 변경 추적    | `df1.compare(df2)`              | 깔끔한 차이만 추출 |
| 전체 row 존재 여부 | `set(map(tuple, df.values))` 방식 | 빠른 비교용     |

1. 패키지 선택

 - 방법: poll하는 패키지의 기반 언어 차이를 통한 성능 향상

 - 상황: broker의 메시지를 감당하지 못해, kafka-python에서 confluent-kafka로 전환

 - 변경점: kafka-python은 python 기반 패키지라 poll한 객체에 대해 property로 바로 접근. confluent-kafka는 librdkafka라는 C기반 라이브러리 기반이기 때문에 C의 객체 반환 방향성에 맞춰 method로 접근. (record.timestamp -> record.timestamp())

2. 구조 측면

이외 parallel consumer라는 패키지도 있으나(역시나 confluent에서 만든 패키지) 기본적으로는 자바 기반이다. (python으로 쓰려면 별도 그냥 쓰레드 구성해야 함)

+ parallel consumer가 만능 키같은 게 아니다. 결국 partition을 늘리기 힘든 브로커 고부하 상황에서, partition 내부까지 병렬 처리함으로써 성능 향상을 꾀하는 건데 안정성, 관리 용이성에 어느 정도 trade-off가 있어 보인다. (https://p-bear.tistory.com/85)

개인적으로 아직까지 보기엔 python 환경에서 구현한다면, docker를 partition 개수만큼 띄우고 병렬 처리하는 게 가장 부작용이 적어 보인다. 이렇게 하면 로컬 테스트도 쉽고 변경이 필요할 때도 구조가 간단한데 병렬 처리는 자연스럽게 가능하다.

그래도 2가지의 불편함&찝찝함이 존재하는데,

1. docker의 리소스 반환이 즉각적이지 않다. 동시에 restart해서 rebalancing하지 않으면, static하게 특정 docker에 과하게 리소스가 부여되는 현상이 관찰되었다. 일반적으로는 먼저 start된 docker에 그런 현상이 나타났다. 물론 이게 docker 형태일 때 피할 수 없는 불편함인가는 아직 잘 모르겠다. (환경은 HyperViser 내의 VM)

2. 지금 다루는 데이터의 크기가 적당히 커서(하루 parquet 기준 500~1000GB 수준) 이 구조가 가장 편한 거 아닐까? 하는 의문은 있다. 이보다 훨씬 커진다면 소스 내에서의 병렬성까지 확보해야 하지 않을까 싶다. 소스내 병렬성까지 생각해야 하는 수준까지 된다면 데이터 순서는 데이터 내부에서 구별해야 하므로, 항상 정확히 정의된 timestamp가 produce부터 고려될 필요가 있다.

이를 바탕으로 내가 겪어본 조합을 떠올려 보면 다음과 같다.

1. raw 데이터가 druid 수준에서 저장된다 => docker 병렬성 가능(druid에서 버티면 웬만큼 가능)

2. raw 데이터가 athena, s3 수준에서 저장된다 + python 환경이다 => docker 병렬성 / 소스내 병렬성 과도기 고민(어쩔 수 없을 때 소스내 병렬성 선택)

3.raw 데이터가 athena, s3 수준에서 저장된다 + python 환경일 필요가 없다 => broker 성능에 따라 미리 parallel consumer를 세팅하는 게 나쁘지 않아 보임

4. raw 데이터가 elastic search, open search, athena, s3, big query 수준에서 저장된다 + python 환경일 필요가 없다. => 거의 parallel consumer는 필수고, partition수를 계획적으로 조정, 테스트할 필요가 있음