데이터 엔지니어링

지난 주 이제는 정말 가족같이 편해져버린 오래된 지인들과 일본여행을 다녀왔다.

그리고 여행에서 “엔저니까..” 라는 이유로 함께 구매한 키보드 세개를 기념샷으로 올려봤다.

9월 부터 열심히 달렸던 프로젝트가 11월 13일에 쉼표를 찍었다. 마침표가 아니라 쉼표다. 주말에도 공부하고, 열심히 일했던 것 같다. 마지막 발표가 있던 주말에는 새벽 3시까지 보고서 PT를 수정했다.

고생했던 순간들을 이 블로그에 남겨서 티도 내고 싶지만.. 빅데이터 워크로드를 쿠버네티스로 이전했던 주제의 후기를 남기고 싶다. (혹시 누군가에게는 도움이 될 것 같다.)

나는 관리형 쿠버네티스 eks위에 emr on eks, airflow on eks 를 구축했는데, 아직까지는 만족 스럽다.

보름정도 지나서 airflow의 한 차례 첫 장애를 맞았지만 해결할 수 있었다.

나의 경우처럼 회사에서 망분리 환경이나 제한된 aws 역할을 갖고 있고, 운영중인 서비스별로 CI/CD 구성이 덜 되있다고 한다면 더 효과를 볼 수 있을지도 모른다. 쿠버네티스의 알려진 장점도 분명이 있다. 하지만 나의 경우에는 결과적으로 각 서버에 터미널로 접속할 일이 줄었고, 워크로드에 따라 스케일 인/아웃, 그리고 서비스 버저닝과 롤백 등의 여러 이점들이 작업시간을 단축해줬다. 앞으로 신규 서비스나 오픈소스를 서버에 올릴 때 EC2를 발급받지 않아도 되고 말이다. 특히 고성능 오토스케일러인 카펜터(karpenter)는 정말 필수적으로 함께 사용해야 한다. Spot,on-demand 인스턴스를 할당할때 놀라운 성능을 보여준다.

단점은 단축된 작업시간이 쿠버네티스 운영과 공부, 트러블 슈팅으로 쓰다보니 아직은 도입 전과 후의 고생의 총양이 같다고 할 수있다. 쿠버네티스 버전 업그레이드 시 데이터 플레인쪽의 수동 업그레이드 작업은 덤이다.

데브옵스 조직이 많이 생겨나고 있는 시대에 데이터 엔지니어가 쿠버네티스 운영 경험이 필요할까? 싶다. 그래도 뭐.. 알아두고 익혀둬서 나쁠 건 없다. 아직은 재미있어서 같이 하고 있다.

앞으로는 짧은 시간동안 몰입했던 여러 기술들을 정리하고 기술 페이지 쪽에 하나하나 포스팅 해보려고 한다.

가족들과 프로젝트 진행에 큰 도움을 준 협력사분들과 부사수에게 감사하다.

앞으로는 적당히 바쁘고 싶다.

친한 선배이자 개발자 동료인 지인이 1일 1글쓰기를 시작 했다며 블로그를 공유해줬다. 그는 교육 강사, 개발자, 또는 회사 대표로서 꾸준히 일한다.

주변 사람에게 선한 영향력을 주고, 그를 보면서 자극을 많이 받아 열심히 노력하게 된다. ( 여기 -> https://www.alghost.co.kr/ )

그래서..   

나 역시 내가 겪는 경험과 생각을 정리하면서, 자유롭게 글을 써보기로 마음먹었다. (마침 창 밖에 빗소리가 내 감성을 채워준다.)

마침 회고할 것이 있다. 가끔 머릿 속 생각을 입밖으로 꺼낼 때가 있다. 그리고 대부분 후회한다.

팀이 하반기 해야 할 일과 방향성을 고민해보고 나눠보는 자리에서, 타부서의 업무지만 우리가 함께 해야 할 것 같아 이야기를 꺼냈다가 싫은 소리를 듣게 됐다.

부서가 담당한 역할과 일이 분명히 구분되있다. 그리고 회사에서는 업무 하나라도 A팀에서 할지 B팀에서 할지.. 담당자들의 보이지 않는 줄다리기를 해야 한다.

난 데이터 엔지니어 입장에서 본다면, 데이터를 수집하는 영역은 우리가 참여해야 할 일이라고 생각했다. 그 데이터는 팀에서 활용 의존도가 높아지고 있다. 우리도 직접 만들면 요청하고 기다리지 않아도 되고, 모두가 좋은거 아닌가 싶었다.

그렇게 오전 회의에서 하지 않아도 될 말을 입 밖으로 꺼내 긁어 부스럼을 만드는 꼴이 됐다. 이런 부분에서 내가 아직 많이 부족하다고 느꼈다. 사실 이전 직장에서도 나의 이런 부분은 고쳐야 한다고 팀장님께서 많이 말씀하셨다. (쉽게 고쳐지지 않지만…)

데이터 엔지니어로서의 나와 빅데이터 플랫폼팀에서 팀원으로서 나의 역할은 같은 것 같지만 다르다. 난 두가지 Role 안에서 신중히 고민하고 의견을 내야 한다.

아직은 R&R 을 따져가며 정치적으로 일하고 싶지 않은 심리가 솔직히 어느정도 있다.

현실적인 것보다 이상적인 것에 더 끌리는 편이다.

내 머릿속 고민의 총량이 있다면, 정치적인 고민보다는 기술적인 고민으로 채우고 싶은 거다.

결론은.. 난 팀장을 하면 안되는 사람이다. 팀원이 피곤해질거니까

1.이슈 내용

- MWAA 에서 실행됐던 snowflake 관련 dag 들이 SnowflakeOperator를 import 할때 에러를 발생시킴

- MWAA  에서 SQLAlchemy 패키지가 최신 업데이트 되면서, 아래 에러를 출력

ile "/usr/local/lib/python3.8/dist-packages/snowflake/sqlalchemy/__init__.py", line 63, in <module>
    from .util import _url as URL
  File "/usr/local/lib/python3.8/dist-packages/snowflake/sqlalchemy/util.py", line 8, in <module>
    from sqlalchemy.engine.url import _rfc_1738_quote
ImportError: cannot import name '_rfc_1738_quote' from 'sqlalchemy.engine.url' (/usr/local/lib/python3.8/dist-packages/sqlalchemy/engine/url.py)```

2.이슈 원인

- sql_alchemy 가 1.4.42버전으로 업데이트 되면서 rfc 1738 즉 quote 처리하는 method 가 rename 되면서 import 하는 코드가 동작하지 않게 되는 이슈가 발생

3.해결 방법

- 일단 급하게 1.4.41 버전으로 다운그레이드 했지만, 버그가 수정된듯 하다.

https://github.com/snowflakedb/snowflake-sqlalchemy/issues/350

https://github.com/sqlalchemy/sqlalchemy/issues/8647

4.회고

- 일단 내가 생각없이 행동했던 조치는 1차 구글링, snowflake 패키지 버전 업데이트 그리고 해결이 안되자 스노우플레이크 기술이사님께 해당 이슈를 공유드리고, 에러가 나는 부분의 코드를 해석 후에 어떤 경우에 발생하는지 코드를 따라가봤다. 

-이슈 공유 드리고, 30분도 안되서 기술 이사님께 github 에 관련 issue 의 링크를 공유 받았고, 다운그레이드로 일단 선 대응할 수 있었다.

- sqlAlchemy 도 오픈 소스인걸 알았음에도 왜 github 에 issue 를 살펴보지 않았을까 후회했다. 

- 습관적으로 구글링에 의존하고, 어떤 설치나 재시작, 재시동 만으로 문제를 편하게 풀려고만 했던 행동들을 반성하자. 

1.이슈 내용

- 파이썬 클래스의 멤버 변수 초기화시 특정 컬럼이 tuple 값으로 초기화되던 문제.

- tuple 로 초기화 된 값 때문에, external 테이블이 미리 정의한 컬럼 타입과 불일치 하여 조회에 이슈가 발생했었다.

2.이슈 원인

-내 손의 잘못.

-콤마. 파이썬은 원소 뒤에 콤마가 있으면 단일 튜플로 인식한다.

- 예제 재연

import json

ods_json_data ='''
{
    "log_seq" : "00001",
    "event_time" : 1656771413
}
'''

class ODSLog:
    def __init__(self,
                 log_Seq = None, 
                 event_time = None,
                 field_1 = None,
                 field_2 = None,
                 **kwargs):
        self.log_seq=log_Seq
        self.event_time=event_time
        self.field_1=field_1,
        self.field_2=field_2
        
def get_ods_transform_log():
    ods_data = json.loads(ods_json_data)
    ods_log = ODSLog(ods_data)
    print(ods_log.__dict__)
   
get_ods_transform_log()
# 출력 값 - 
{'log_seq': {'log_seq': '00001', 'event_time': 1656771413}, 'event_time': None, 'field_1': None, 'field_2': (None,)}

- ods_json_data 와 같은 방식으로 데이터가 들어오면, 클래스를 이용하여 필요한 ETL 처리후에 데이터를 출력 처리

- 실시간 데이터 수집의 경우 이상값이나 정의하지 않은 컬럼이 들어올 때를 예측해서 처리 로직을 추가해주면 좋다.

3.회고

콤마가 포함된 라인 하나를 발견하지 못해, 원천 데이터를 탓하며 일부러 타입 캐스팅을 하드코딩할 생각까지 했던 스스로를 반성한다. 

컬럼이 엄청 많은 로그를 처리할때 sublime text 나 엑셀을 이용해 프로그램 코드의 일부분을 일괄로 처리할 때가 있다. 그때 일부는 하나하나 찾아가면서 보정을 하게 되는데 그 과정에서 실수가 많이 발생하는 것 같다.

하반기에는 파이썬 공부도 해야 할 것 같다....