새소식
반응형
Data pipeline/ETL
데이터 파이프라인 제작기 - 반정형 데이터편 (JSON)
- -
반응형
안녕하세요.
이 내용을 쓸지 말지 정말 고민을 많이 했습니다.
예전에 링크드인에 관련하여 포트폴리오 형식으로 PPT를 만들어서 올렸는데요.
조회수가 3만에 육박하여 굉장히 많은 분들이 관심을 가져 주셨습니다.
그래서 매번 써야지 써야지 했는데 내용이 워낙 길어서 고민을 많이 했습니다.
간단하게 쓸 수 없었습니다...
1.0 개발 - 2개월
2.0 개발 - 2개월
3.0 개발 - 2개월
이렇게 총 3번의 버전 업데이트를 하였습니다.
처음에는 데이터를 볼 수 없어서 분석을 하지 못하였는데, BigQuery로 넣는 작업부터 시작하여...
재사용성을 고려하여 코드를 리팩토링 하였고, 마지막으로 multi thread를 적용하였습니다.
코딩을 2019년에 처음 시작하여, 지금의 회사에서 팀장님과 팀원들에게 많이 배우면서 클린 코드와 여러 가지 성능에 관련하여 배우다 보니 N차 개발할 때마다 거의 새로 만드는 수준으로 만들었습니다.
이 글에서는 코드는 공개하지 않으며 (부끄럽습니다.) 만들면서 고민하였던 부분과 해결하였던 부분을 다루고자 합니다. 그리고 내용은 1탄(반정형) -> 2탄(정형)으로 할 예정입니다. 제가 비정형 데이터를 다루어 본 적은 없기 때문에 이렇게 2개만 올리고자 합니다.
1. 어디서 어디로 가는가?
출발지 -> S3 (AWS)
목적지 -> BigQuery (GCP)
출발지를 S3로 한 이유는 앞에 부분에 Elastic search(이하 ES)가 있기 때문입니다.
ES에서 S3로 데이터를 몇 MB당 1회 또는 몇 분당 1회씩 데이터를 S3에 파일 형식(JSON)으로 저장할 수 있습니다.
그래서 S3에 저장된 데이터를 출발지로 합니다.
(기존의 세팅이 ES였고, S3에 이미 데이터가 저장되고 있었습니다.)
목적지는 BigQuery입니다. 제가 재직 중인 회사에서는 DataLake / DataMart를 BigQuery에 저장하고 있습니다. 감히 말씀드리지만 BigQuery는 정말 강력하고 편한 서비스입니다. (제가 써본 Bigdata 관련 서비스 중에선...)
🤷♂️ Why?
Q. 왜 AWS , GCP를 섞어서 사용하였나요?
| 굳이 왜? 그렇게 했나요? 하고 궁금하신 분들이 많을 듯 합니다. 사실 이 부분은 제 선택은 아니였습니다. 2021년 현재 재직중인 회사의 경우 Data Lake 및 Mart 모두 BigQuery에 저장하고 있습니다. 로그의 경우 ES에서 받아서 S3에 저장하고 있었습니다. 그렇기 때문에 제가 이미 하고 있는 시스템을 모두 바꾸는 것보다는 유연하게 기존의 있는 데이터를 어떻게 사용자가 쓰기 편한 곳으로 ETL 해주는 것이 더 중요하다고 생각 되었습니다. |
2. 전체적인 데이터의 흐름
위에서 언급한 출발지와 목적지는 S3와 BigQuery입니다.
이것을 하나씩 그림으로 데이터의 흐름을 그름으로 나타냈습니다.
생각보다 간단합니다. 몇 곳을 거치지 않습니다. (사실 전부 python code...)
데이터의 흐름에 대해서 간략하게 말씀드리면 아래의 과정을 거칩니다.
데이터는 분류기, 적재기로 2가지 공정(?)을 거쳐서 출발지에서 목적지로 들어갑니다.
2.1 분류기
1. ES에서 S3로 YYYY/MM/DD/HH로 폴더 화하여 JSON 데이터를 일정 간격, 크기로 저장합니다.
2. S3의 저장된 JSON의 경우 로그의 타입이 Key로 있지만 파일에 모두 모여서 들어가 있습니다.
- /n으로 구분자가 있으나, 모든 로그 타입이 한 JSON파일에 있습니다.
3. 우선 JSON 파일을 kubernetes pod의 local에 내려받습니다.
4. 이 파일을 열어서 각 타입별로 데이터를 나눠서 각각의 파일로 만듭니다.
5. 만들어진 각각의 파일을 GCS Bucket으로 로그 타입/YYYY/MM/DD/HH로 업로드합니다.
🤷♂️ Why?
Q. 왜 S3에서 Kubernetes pod의 disk에 다운로드하여서 분류 하나요?
| 처음에는 S3에서 바로 읽어서 GCS에 분류하면서 적재를 하려고 하였습니다. 하지만 S3에서 JSON데이터를 읽는 것은 가능하지만 multi thread로 데이터를 GCS Bucket에 쓰는 방법을 찾지 못하였습니다. 그래서 우선 Local에 받아서 열어서 multi thread를 통해서 데이터를 빠르게 쓰고 이를 다시 GCS에 업로드 하는 방법을 사용 했습니다. |
🤷♂️ Why?
Q. 왜 로그를 굳이 분류하였나요?
| 로그를 한번에 넣으려면 스키마가 맞지 않아서 BigQuery에서는 "Too Many Error를 뱉습니다...(정말 친절한) 그래서 스키마를 맞춰서 넣어줘야 합니다. 많은 분들이 그럼 String으로 묶거나 안쪽에 또 다시 JSON으로 묶는 방법을 고민해보지 않았는지 묻습니다. 하지만 그렇게 하면 사용자 입장에서 또 한번 JSON을 Parsing해야 합니다. 제가 의뢰 받은 로그의 경우 스키마가 거의 변경이 없었기에 스키마를 고정적으로 만들어놓고 로그를 타입별로 분류하여 각각 틀(스키마)에 맞게 데이터를 적재 하는 것이 가능 하였습니다. |
2.2 적재기
사실 적재기는 큰 기능은 없습니다.
이 링크를 통해서 Python을 개발해서 GCS에 저장된 데이터를 그대로 BigQuery에 넣었습니다.
제가 실제로 넣은 옵션을 보면 다음과 같습니다.
|
print("___________________________________________________________________")
print(f"{self.log_type} schema \n")
pprint.pprint(json_schema)
print("___________________________________________________________________\n")
job_config = bigquery.LoadJobConfig()
job_config.schema = json_schema
job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
uri = f"gs://{self.gcs_bukcet_name}/{self.file_name_prefix}*"
load_job = self.client.load_table_from_uri(
uri, self.dataset_ref.table(f"{self.log_type}_TEMP"), job_config=job_config
)
load_job.result()
self.client.get_table(self.dataset_ref.table(f"{self.log_type}_TEMP"))
|
cs |
위의 부분에서 제가 생각하기에 가장 중요한 부분은 job_config.schema = json_schema입니다.
코드의 재상용 성을 위해서 JSON 파일로 스키마를 저장하고 이를 불러와서 처리하였습니다.
기존에 1.0 버전에서는 각 로그 타입마다 각각 코드를 만들어서 이 스키마 부분만 변경하여 코드를 만들었습니다.
하지만 매우 불필요하고 재사용성이 떨어져서 JSON 형태로 스키마를 불러와서 적재하는 방법으로 변경하였습니다.
또한 위의 코드를 살펴보시면 데이터를 데이터를 실제 테이블이 아닌 TEMP에 먼저 넣습니다.
이는 로그의 적재 시간을 함께 기록하기 위함입니다.
" 로그의 적재 시간은 이미 데이터 안에 있지 않아? "
네! 맞습니다. 로그의 적재 시간은 데이터에 있습니다. 하지만 제가 말하는 적재는 로그가 쌓인 시간이 아닙니다.
S3 Bucket 쌓인 시간을 뜻 합니다.
내가 만약에 00시 ~ 01시까지의 데이터를 적재한다고 하여, 정확하게 그 데이터만 있지 않습니다.
S3에서 실제로 JSON 데이터를 열어보면 마지막 데이터의 경우 01:01~05분 이런 식으로 데이터가 뒤에 더 들어옵니다.
이유는 ES에서 몇 분당 또는 몇 MB당 데이터를 S3로 보내는데, 아직 해당 용량이 차지 않아서 해당 시간에 오는 경우입니다.
말로 설명하는 것보다 그림으로 보는 것이 더욱 빠를 듯합니다.
눈치가 빠르신 분들은 위의 그림만 봐도 무엇인지 이해가 가실 듯합니다.
JSON 데이터의 있는 시간을 믿고 데이터를 적재할 경우 문제가 되는 경우는 재적재 이슈가 있을 경우 있습니다.
데이터를 지우고 다시 넣어야 하는데, 지울 때는 로그에 있는 날짜를 쓰기 때문에 뒤에 더 들어온 데이터는 지워지지 않습니다.
그래서 데이터 기준으로 데이터를 지우는 게 아니고, 폴더 기준으로 데이터를 지워야 정확하게 재적 재가 가능합니다. 이를 멱등성이라고 하며, 몇 번을 지워도 같은 결과가 나와야 합니다.
위에서 TEMP 테이블에 넣고 폴더 시간을 같이 넣어서 원본 테이블에 INSERT 하는 이유는 이런 이유 때문입니다.
🙋♂️ TIP
TEMP 테이블에 데이터를 먼저 넣으면 좋은 것
| S3에 데이터의 경우 기본적으로 UTC로 쌓였습니다. 그래서 이러한 데이터를 사용자가 보기 편하도록 KST로 변환해주는 작업도 할 수 있습니다. 저 같은 경우 TEMP에서 Origin으로 넣을때 INSERT ... SELECT 구문을 사용 하였는데요. 이때 Query에서 여러가지 작업을 해서 일부 데이터를 가공해서 넣어줬습니다. (물론 요청도 있었습니다.) 또한 데이터가 TIMESTAMP 형태로 숫자로 들어와서 이를 날짜 형태로 바꾸는 작업이라던가 여러가지 작업을 해줄 수 있었습니다. |
3. 데이터 스키마 잡기 및 QA 방법
이 부분은 전에 링크드인 글에서 다루지 않은 내용입니다.
그래서 조금 더 많은 정보를 공유하고자 어떻게 데이터를 QA 하였는지 작성하였습니다.
사실 크게 거창한 부분은 없습니다.
그리고 생각해보니 이것은 데이터 QA라기보다는 데이터 스키마 QA에 관점이 더 맞는 것 같습니다.
그런데 이 부분이 제일 중요하다고 가장 많이 느낍니다. 데이터를 어떻게 쌓고 어떻게 볼지를 합의하지 않고 무조건 넣는 것에 초점을 맞추면 나중에 파이프라인을 만들고 엄청나게 여러 번 수정을 할 수 있습니다. (얼마나 중요하면 저렇게 쓰겠습니까..)
python code에서 if... if... if... if 하고 여기서 끝나면 다행이겠죠? BigQuery의 경우 스키마의 Key와 자료형이 바뀌면 안 들어갑니다. 이때도 역시 우리의 "Too Many Error" 일으키며 데이터가 들어가지 않습니다.
🤷♂️ Why?
Q. BigQuery 적재 시 오류 무시 기능 (max_bad_recorde)을 쓰지 않는 이유는?
| 위에서 많은 row가 아닌 몇건의 row의 경우 BigQuery에서 데이터를 넣을때 max_bad_recorde기능을 사용하면 스키마가 맞지 않는 일부의 row를 무시하고 문제없이 적재 할 수 있습니다. 하지만 이 기능을 사용하지 않았습니다. 만약에 이렇게 몇건의 row를 계속해서 쌓지 않았는데, 이 row가 실제 필요한 데이터이거나 혹은 우리 서버나 개발에 치명적인 로그인 경우 당연히 문제가 될 수 있으므로 모든 데이터를 쌓고자 하였습니다. |
그럼 또 저는 열심히 데이터를 분류하고 적재하고.. 이 작업을 해줘야 합니다.
1~2달 정도는 괜찮습니다. 그런데, 갑자기 1년 치 데이터를 그렇게...? 정말 까마득합니다.
3.1 개발자와 많이 이야기한다.
개발자와 많이 이야기했습니다.
개발자 즉, 로그를 심어주신 분을 뜻 합니다. 로그를 제가 심지는 않습니다.
물론 제가 심을 수도 있지만 담당자가 있으니 잘하는 사람에게 맡기는 게 맞지 않을까요...?
그래서 개발자와 최대한 많이 이야기했습니다. 이것은 무엇이고 저것은 무엇이고, 이건 왜 이렇게 오는지 저건 왜 저렇게 오는지?
로그를 심는 사람은 데이터를 어떻게 적재하면 최대한 예쁘게 쌓이는지 보다는 잘 만들지를 고민합니다.
그래서 쌓아야 하는 본인 (나)이 이 부분에 대해서 이야기를 많이 해야 한다고 생각합니다.
3.2 사용자와 많이 이야기한다.
뻔한 이야기입니다. 하지만 이것도 중요합니다.
"이 때 말하는 사용자는 로그를 적재해달라고 요청한 사람입니다."
적재를 해달라고 요청 한 사람이 심어주는 사람과의 생각과 100% 일치할 순 없습니다.
그래서 데이터를 심는 사람과 쓰는 사람의 중간에서 역할이 중요합니다.
조율이 필요합니다. 데이터를 보고 중간의 합의점을 찾아서 조율해줬습니다.
예를 들면 심는 사람의 입장에서는 String으로 모두 넣으면 정말 편합니다.
하지만 쓰는 사람 입장에서는 당연히 안됩니다. SUM , AVG 등 함수를 사용하면 String을 변환해야 합니다.
또한 그렇게 넣었다가 무엇은 INT인데... 무엇은 String이다? 이러면 정말 곤란합니다. 🙅♂️
3.3 합의점을 찾은 뒤 적재하면서 QA 방법
위의 2가지 방법으로 합의점을 찾고 로그를 적재하기 시작하였다면, 이제 실제로 적재한 뒤 데이터가 맞는지 봐야 합니다.
저는 다음의 방법을 통해서 데이터가 맞는지 확인하였습니다.
" 일단 적재한다. "
데이터를 적재합니다.
분류된 데이터를 하루 또는 1시간 정도 테이블 만들기를 통해서 만듭니다.
여기서 꼭 스키마 옵션에서 "자동 감지"를 선택해 주세요.
자동 감지를 통해서 스키마가 맞는지 어느 정도 확인 가능합니다.
여기서 또다시... "Too Many Error"를 보신다면 데이터가 맞는지 다시 한번 확인해야 합니다.
BigQuery의 경우 어디서 무엇이 틀렸는지 친절하게 설명해 주지 않습니다... (정말 힘듭니다.)
보통 위에서 합의를 봤다면 문제없이 들어갑니다.
그렇기 때문에 만들어진 테이블 이용해서 다음의 글을 확인해서 스키마를 추출할 수 있습니다.
https://burning-dba.tistory.com/143?category=1027244
BigQuery - 테이블에서 스키마 추출
BigQuery의 다수의 테이블의 스키마를 Json으로 봐야 하는 작업이 있어서 한 땀 한 땀 열심히 만다가 문득 스키마를 추출해서 파일로 저장할 수 없을까?라는 생각이 들어서 찾아봤습니다. 🙋♂️
burning-dba.tistory.com
스키마를 왜 추출할까요?
위에서 언급한 JSON파일을 이용한 스키마 저장을 위함입니다.
JSON 형태로 스키마를 추출해서 나중에 해당 파일을 불러와서 TEMP 테이블을 만들기 위함입니다.
위의 내용을 어느 정도만 해줘도 데이터의 스키마가 틀려서 적재가 실패하는 경우는 거의 없었습니다. (아예 없었습니다.)
4. 파이프라인 이모저모
이 부분은 조금 ETC 느낌으로 쓸 것 같습니다.
ETC가 좋은 것은 아니지만 한 번에 묶어서 쓰기에는 이 파트의 제목을 무엇이라고 할지도 고민을 하였습니다.
그리고 이걸 세부적으로 나누면 너무 길어질 것 같아서 생각나는 부분을 적어놨습니다.
만들면서 겪었던 문제나 해결방법 또는 이런 부분을 알면 좋을 것 같은 이야기를 적어놨습니다.
4.1 적재 주기
적재의 경우 Airflow를 이용해서 배치 처리하였습니다.
이 부분도 관심이 많으실 것 같습니다.
데이터를 실시간으로 하는지 배치 처리를 하는지 이 부분은 제가 데이터를 하면서 많은 분들이 중요하게 생각하였습니다.
우선 제가 만든 이 파이프라인은 1시간 단위로 적재를 합니다.
하지만 여기서 중요한 부분이 있습니다. 데이터의 적재를 딱 정각에 시작하지 않았습니다.
여기에는 중요한 이유가 있습니다.
ES에서 S3로 데이터를 저장할 때 바로바로 저장하지 않는다고 하였습니다.
조금씩 늦게 들어오는 경우가 존재한다고 위에서 언급하였습니다.
그래서 데이터를 가져올 때 논리적으로 조금의 텀을 두고 배치 처리하였습니다.
그림으로 보면 더욱 이해가 빠를 거 같아서 그림으로 준비하였습니다.
위의 데이터 적재의 경우 적재를 나타내며, 분류의 경우 -1 시간 전 데이터를 가져옵니다.
위의 데이터의 경우도 -1 시간 전의 데이터를 기준으로 합니다.
ES에서 S3에 데이터를 모두 넣었다고 가정하고 00시 데이터를 적재한다고 가정한다면, 다음과 같은 스태프로 데이터를 처리합니다.
1. 01시 05분에 00시 데이터가 YYYY/MM/DD/00 폴더에 모두 적재를 마친다.
2. 00 폴더의 데이터를 읽어서 분류하여, GCS에 적재한다. (분류의 경우 피크타임 기준으로 5분을 넘지 않습니다.)
3. 혹시 모를 지연을 대비하여, 10분 정도 기다렸다가 시작하는 Sensor를 걸어둔다.
4. Sensor에서 분류기가 끝났는지 확인 후 빠른 적재 처리를 시작한다.
5. 나머지 빠른 적재 이외의 일반 적재는 30분~정각 사이에 분산적으로 처리한다.
- 여기에서 분산은 Airflow의 스케줄 시간을 분산시켰음을 의미합니다.
- Kubernetes 환경이지만 자원이 무한이 아니고 Dag이 매우 많기 때문에 1~2분 간격으로 나누어서 적재하도록 하였습니다.
- kubernetes의 경우 100m , 128MB를 기준으로 pod가 한 개 뜨는데, 이 처리를 위해서 100개 정도의 pod가 실행됩니다. 이 데이터만 처리하기 위해서 세팅된 것이 아니기 때문에 조금씩 분산해서 스케줄 처리를 하였습니다.
🙋♂️ TIP
위에서 언급한 Airflow의 Sensor 스케줄 시간을 세팅할 때 이걸 꼭 참고하세요.
| 위에서 언급한 Sensor의 경우 스케줄 시간이 기존의 Dag과 일치 하지 않을 경우 무한으로 기다릴 수 있습니다. 그래서 스케줄 시간(Dag 실행 시간)은 같게 세팅하고 timedelta를 통해서 더하거나 뺴줘야 합니다. |
코드를 참고하시면 더욱 좋을 것 같습니다.
|
sensor = ExternalTaskSensor(
task_id="wait_task_1",
external_dag_id="log_classification",
external_task_id="t1",
execution_date_fn=lambda x: x - timedelta(minutes=5),
mode="reschedule",
timeout=3600,
)
|
cs |
4.2 어쩔 수 없이 스키마가 계속 바뀐다면?
사람 앞길은 알 수가 없듯이 데이터도 비슷하다고 봅니다.
위에서 스키마가 다르면 데이터를 적재할 때 "Too Many Error"가 나온다고 하였습니다.
그런데, 생각해보면 데이터가 항상 꼭 같을 수 있을까요?
제가 적재를 하다가 이런 경우가 있었습니다.
" 타사의 API 응답을 가져와서 우리 쪽 로그로 받아와야 해서 형태가 자꾸 변합니다.
또한 어떻게 들어올지 알 수 없어요... "
사실 알 수는 있겠지만 (문서를 받거나 계속 알아본다면?)
어쩔 수 없는 상황이 발생하기도 합니다. 그래서 가변적으로 스키마를 받아야 할 경우 다음과 같이 해결하였습니다.
JSON이지만 string으로 묶습니다.
양쪽에 "를 추가해서 JSON이지만 string으로 묶습니다.
그리고 BigQuery에서 Parsing 해서 사용하였습니다.
https://burning-dba.tistory.com/142
BigQuery - JSON 컬럼 파싱하기
안녕하세요. BigQuery를 사용하다 보면 자주 접하는 Column이 있습니다. 바로 JSON으로 만들어진 Column입니다. 데이터의 형태가 계속해서 변하거나 항목들이 가변적으로 들어오는 경우 JSON 형태로 데
burning-dba.tistory.com
위의 링크를 참고해서 값을 불러서 파싱 해서 사용할 수 있습니다.
🤷♂️ Why?
Q. 이럴 거면 스키마를 모두 위오 같이 만들지 왜 굳이 스키마를 잡아줬나요?
| 처음에 그렇게 하자는 의견도 있었습니다. 그런데 스키마를 잡아두면 여러가지 이점이 있다고 생각 됩니다. 첫번째, 사용자가 쓰기 편합니다. 당연하겠지만 JSON으로 parsing하지 않고 바로 쓰기가 가능 합니다. 두번째, 명확하게 데이터가 들어오는지 볼 수 있습니다. JSON으로 모두 String으로 넣으면 그 데이터가 맞는지 틀린지 다시 parsing해서 검사를 해야합니다. 세번째, 나중에 한눈에 데이터의 형태(테이블 형태)를 볼 수 있습니다. 스키마를 잘 정리해두면 어떻게 생겼는지 한번에 알기 편합니다. 위와 같은 여러가지 이유로 인해서 일부를 제외한 스키마를 잡아서 형태를 만들어줬습니다. |
4.3 데이터를 명확하게 잡고 적재를 고민하자.
데이터 엔지니어를 하면서 많은 팀과 많은 분들과 이야기를 나누었습니다.
전부는 아니지만 많은 분들이 데이터를 어떻게 효율적이고 빠르게 ETL 할지를 고민합니다.
그런데, 제가 생각에 가장 중요한 것은 데이터를 명확하게 잡는 게 중요하다고 봅니다.
4.3.1 0과 null은 다르다.
Key값에 데이터가 안 들어올 경우 0으로 넣었다고 가정을 합니다.
그렇다면 이 값은 0이 맞을까요? 제가 생각하기엔 아닌 것 같습니다.
값이 들어오지 않았으므로 null로 찍어야 맞습니다.
값이 안 들어와서 아예 없는 것과 0은 완전히 다른 이야기입니다.
위의 데이터에서 문제는 이렇습니다.
우리가 어떤 데이터를 쌓는데, API가 오류가 있어서 특정 key의 값이 들어오지 못했습니다.
들어오지 못했으므로 이걸 0으로 찍었습니다. 그럼 사용자 입장에서는 이건 들어온 값입니다.
0으로 값이 들어왔습니다. 그래서 이 값은 잘못된 값인지 실제로 0이라는 값이 들어온 것인지 알 수 없습니다.
4.3.2 데이터를 모두 String으로!?
데이터를 심으면서 가끔 데이터가 안 들어간다는 이유로 String으로 통일해서 넣는 경우가 있습니다.
이렇게 String으로 모두 통일해서 넣으면 당연히 잘 들어갑니다.
그런데, 문제가 하나 발생합니다.
SUM , AVG와 같은 함수를 쓸 때 데이터가 String이면 CAST(value AS INT64)와 같은 구문을 이용해서 변환을 한번 해줘야 합니다.
이렇게 하면 일단 Query의 양이 늘어나고 매번 번거롭게 처리가 필요합니다.
그리고 실수로 (그럴리는 없겠지만..) 해당 값에 INT가 아닌 실제 String이 있으면...!?
SELECT가 안됩니다. 그럼 또 열심히 찾아서 보정해줘야 합니다.
그렇기 때문에 모든 데이터를 String으로 맞춰서 넣는 것은 좋은 방법은 아니라고 봅니다.
5. 모니터링
모니터링도 상당히 신경 쓴 부분 중에 하나입니다.
여러 가지 이유에서 모니터링을 하였지만 데이터의 신뢰성을 위해서 제일 중요하다고 생각합니다.
모니터링의 경우 Slack , DataStudio를 통해서 만들었습니다.
Slack을 통해서 매일 몇 건의 데이터가 적재되었는지 알람을 보냈고, DataStudio의 지표를 통해서 데이터의 적재 추세를 보았습니다.
데이터의 적재가 갑자기 줄거나 데이터의 적재가 너무 많을 경우 중복이나 누락을 의심하여 조치를 하였습니다.
5.1 Slack 모니터링
Slack으로 표를 만들어서 해당 로그 타입이 몇 건이 적재되었는지 보냈습니다.
(표 형태로 출력이 없어서 직접 Python으로 만들어봤습니다. / 은근히 물어보는 지인이 많으십니다..😂 )
사내에서 사용하는 메신저가 Slack이라서 선택 한 부분이 가장 컸습니다.
막상 보내보니 굉장히 보기도 편하고 좋았습니다.
5.2DataStudio 지표
날짜 별로 데이터의 적재 수치를 표현하여 몇 건이 적재되었는지 지표화 하였습니다.
추세를 보면서 언제 어떠한 로그가 들어온 지 보면 어느 정도 예측도 가능 하였습니다.
생각보다 로그 타입의 건수를 보고도 어느정도 예측이 가능합니다.
예를 들어서 매출에 관련된 구매 타입의 로그가 건수가 많을 경우 오늘은 매출이 조금 더 나오겠다? (물론 금액은...)
이런 정도의 유추가 가능합니다.
실제로 저 로그의 추세를 보고 장애 여부를 판단한 경우도 있었습니다.
(특정 로그가 급격하게 조금 들어와서)
6. 마치며...
아직도 수정하고 업데이트할 내용이 굉장히 많습니다.
일단 쓸 수는 있게 만들었다(?) 정도로 만든 것 같습니다.
하면서 많은 분들의 도움을 받았고 감사하게도 많은 부분을 세심하게 알려주셨습니다.
그리고 몇 가지 개선할 부분이 있는데 제일 먼저 하고 싶은 부분은 전부 Paas에 의존하고자 합니다.
지금 보시면 Airflow를 통해서 배치 처리를 하고 있습니다.
Airflow의 경우 위에서도 언급하였지만 Kubernetes위에 있고 자원도 한정적입니다.
그렇기 때문에 AWS의 SQS를 이용해서 파일이 S3에 생기자마자 분산 처리 후 lamdba를 통해서 바로 적재를 하는 방향으로 추가 개발을 하고자 합니다. 완전히 Paas로 만들어서 모든 의존성을 없애는 게 긍극적인 목표입니다.
그리고 제가 데이터 엔지니어가 된 지 얼마 안 됐을 때 개발한 내용이라서 부족한 부분이 굉장히 많습니다.
아마도 왜? 굳이?라는 의문을 가지시는 분들도 많을 듯합니다.
하지만 "이러한 방법도 있다"라는 생각으로 가볍게 봐주셨길 바랍니다.
긴 글 읽어 주셔서 감사합니다.
반응형
'Data pipeline > ETL' 카테고리의 다른 글
| [ETL] RDB에서 데이터 ETL을 위한 최소한의 테이블 설계 (0) | 2023.02.02 |
|---|---|
| [공통] 데이터를 적재하고 보기까지 (2) | 2022.06.16 |
| RDBMS 데이터 적재 시 데이터 정합성 체크 (0) | 2021.04.15 |
| 데이터 파인프라인 제작기 (0) | 2021.03.16 |
Contents
소중한 공감 감사합니다