LabHub
배우기 러닝패스 코스

Data Pipelines

Yesterday's Number Changed Today: Watermarks and Late Data

LabHub 에서 이어서 보기

한국어 원문으로 표시합니다.

목표

이벤트 시간으로 집계하는 도구 wm.py 를 만든다. 처리 시간과 이벤트 시간의 벌어짐을 분포로 재고, 워터마크를 도착 순서대로 움직이고, 창이 닫힌 뒤 온 자료를 가르고, 버리는 정책과 반영하는 정책의 결과를 견주고, 닫는 순간의 값과 그 뒤의 정정을 따로 남긴다.

왜 중요한가

이 코스의 앞 실습에서 워터마크를 한 번 다뤘다. 적재한 자료의 최대 시각을 표에 적어 다음 실행이 그 뒤만 읽게 하는 것이었고, 그 워터마크가 답하는 질문은 어디까지 읽었는가 하나였다. 여기서는 그 다음을 다룬다. 모바일 앱은 신호를 잃었다가 몇 분 뒤에, 비행기 모드였던 단말은 몇 시간 뒤에 이벤트를 보낸다. 그 이벤트가 일어난 시각은 이미 지나간 시각이고, 우리는 그 시간대의 집계를 이미 내보냈다. 워터마크는 그 자료가 더 오지 않는다고 본다는 약속이지 사실이 아니다. 그래서 정할 것이 셋이다. 허용 지연을 얼마로 잡을 것인가, 약속을 어기고 온 자료를 버릴 것인가 늦은 갱신으로 반영할 것인가, 그리고 반영했다면 어제 낸 숫자와 오늘 낸 숫자의 차이를 어떻게 설명할 것인가. 시간을 재는 실습이 아니다. 이벤트에 적힌 event_time 과 ingest_time 이라는 두 칸으로 계산하는 실습이고, 프로그램이 실제로 몇 초 걸리는지는 아무 상관이 없다. 채점기는 여러분이 적어 낸 문구를 믿지 않는다. 임시 파일에 채점기가 만든 이벤트 흐름을 차려 놓고 여러분의 도구를 실제로 돌려, 창 크기와 허용 지연을 바꿔 가며 답을 대조한다. 건수와 지연 분포는 실행마다 바뀐다.

단계

  1. /root/wmark/gen_stream.py 를 만들어 실행해 /root/wmark/work/stream.jsonl 을 만드세요.
  2. /root/wmark/wm.py 에 skew 를 만들어 지연 분포와 순서가 어긋난 건수를 내게 하세요.
  3. watermark 를 더해 도착 순서대로 워터마크를 움직이게 하세요.
  4. windows 를 더해 이벤트 시간으로 창을 나눠 집계하게 하세요.
  5. late 를 더해 창이 닫힌 뒤 온 자료를 가르게 하세요.
  6. agg 를 더해 버리는 정책과 반영하는 정책의 결과를 각각 내게 하세요.
  7. close 를 더해 닫는 순간의 값과 그 뒤의 정정을 따로 내게 하세요.
  8. /root/wmark/work/watermark_report.json 과 /root/wmark/work/watermark_report.md 를 쓰세요.

참고

늦게 오는 이벤트를 손에 쥐기

/root/wmark/gen_stream.py 를 만들어 실행해 /root/wmark/work/stream.jsonl 을 만드세요. 60건 이상이고, 파일 차례가 ingest_time 오름차순이며, 모든 줄에서 ingest_time 이 event_time 이상이고, 지연이 120초 이상인 줄이 5건 이상, 이벤트 시간의 폭이 1200초 이상, key 는 3가지 이상이어야 합니다.

지연을 한 가지 분포로만 만들면 뒤에서 볼 것이 없습니다. 대부분은 몇 초 안에 들어오고, 일부는 몇 분, 아주 일부는 수십 분 뒤에 밀려드는 식으로 세 갈래로 흩뿌리세요. 다 만든 뒤 ingest_time 으로 정렬해 파일에 쓰면 그것이 도착 차례가 됩니다. 씨앗을 고정해야 허용 지연을 바꿔 가며 견주는 동안 자료가 흔들리지 않습니다.

얼마나 벌어지는지부터 잰다

/root/wmark/wm.py 에 skew <파일> 을 만들어 건수·순서가 어긋난 건수·지연의 최소·최대·중앙값·95분위를 JSON 으로 내게 하세요.

지연은 ingest_time - event_time 입니다. 분위수는 가장 가까운 순위로 잡고 보간하지 마세요 — 정수로 떨어져야 채점도 회의도 흔들리지 않습니다. 순서가 어긋난 건수는 도착 차례대로 훑으며 지금까지의 최대 이벤트 시간보다 이른 것을 세면 됩니다.

워터마크를 도착 순서대로 움직이기

watermark <파일> --lateness=<초> 를 더해 lateness·max_event_time·advances·final_watermark 를 내게 하세요. advances 는 최대 이벤트 시간을 새로 갱신한 도착 건수입니다.

워터마크는 뒤로 가지 않습니다. 늦게 온 이벤트가 최대 이벤트 시간을 낮추게 두면 이미 닫은 창이 다시 열렸다 닫히기를 반복하고 어떤 값도 확정되지 않습니다. 첫 도착은 비교할 앞 자료가 없으므로 그 자체로 한 번의 전진입니다.

이벤트 시간으로 창 나누기

windows <파일> --size=<초> 를 더해 창마다 건수와 금액을 내게 하세요. 창 시작은 event_time - (event_time % size) 이고, 이 단계에서는 늦고 이르고를 따지지 않고 전부 셉니다.

JSON 의 키는 문자열이어야 하므로 창 시작을 문자열로 적습니다. 나중에 정렬할 때는 문자열이 아니라 정수로 견주세요 — 자릿수가 다르면 문자열 정렬이 엉뚱한 차례를 냅니다.

창이 닫힌 뒤 온 자료 가려내기

late <파일> --size=<초> --lateness=<초> 를 더해 on_time·late·late_by_window 를 내게 하세요. 어느 이벤트가 늦었는지는 그보다 먼저 도착한 것들로만 계산한 워터마크로 판단하고, 워터마크가 그 창의 끝 이상이면 늦은 것입니다.

늦음은 자료의 성질이 아니라 도착 차례의 성질입니다. 다루기 편하라고 이벤트 시간으로 한 번 정렬해 버리면 도착 차례가 사라지고 늦은 자료가 0건으로 나옵니다 — 문제가 없어서가 아니라 볼 눈을 없앤 것입니다. 파일에 적힌 차례 그대로 훑되, 최대 이벤트 시간 갱신은 판정을 마친 뒤에 하세요. 첫 도착은 비교할 앞 자료가 없으므로 늦지 않습니다.

버릴 것인가 반영할 것인가

agg <파일> --size --lateness --policy=drop|update 를 더하세요. drop 이면 늦은 자료를 빼고 dropped 로 세며 restated 는 빈 목록이고, update 면 늦은 자료도 넣고 dropped 는 0 이며 늦은 자료가 들어간 창을 restated 에 담습니다.

버리는 쪽을 고르더라도 버린 건수를 반드시 세세요. 세지 않고 버리면 나중에 합계가 안 맞는 이유를 설명할 근거가 없습니다. restated 는 창 시작을 오름차순으로 담되 견줄 때는 정수로 견주세요. 모르는 정책 이름이 오면 종료 코드 2 로 끝냅니다.

닫는 순간의 값과 정정을 따로 남기기

close <파일> --size --lateness 를 더해 sealed(닫히기 전에 온 것만)·corrections(창 시작 오름차순의 정정 목록)·final(둘을 더한 값)을 내게 하세요. 늦은 자료만 있는 창의 sealed 는 0 입니다.

둘을 합쳐 하나로만 내면 어제 숫자가 오늘 바뀐 이유를 설명할 길이 없습니다. 따로 남기면 바뀐 양이 곧 대답이 됩니다. corrections 에는 정정이 실제로 있는 창만 담고, final 은 모든 창을 담습니다.

바뀐 숫자를 설명하는 한 장

/root/wmark/work/watermark_report.json 에 size·lateness·events·windows·on_time·late·dropped·restated·max_lag·p95_lag·policy 를 적고, /root/wmark/work/watermark_report.md 에 ## 무엇을 재었나 ## 허용 지연을 얼마로 잡았나 ## 늦게 온 자료를 어떻게 했나 ## 어제 숫자가 바뀐 이유 네 절로 쓰세요.

허용 지연은 1 이상 최대 지연 미만으로 잡고, 그 값에서 늦게 온 자료가 1건 이상 나와야 합니다. 나오지 않으면 값을 줄이세요. 보고서에는 늦게 온 건수를 숫자로 적으세요 — 그 숫자 하나가 다음 회의에서 가장 먼저 나오는 질문의 답입니다. 앞 단계에서 만든 함수를 그대로 부르면 됩니다.