data/2021/06/01/data-engineering-with-airflow.html

[socar-kyle]

쏘카 데이터 그룹 - Airflow와 함께한 데이터 환경 구축기(feat. Airflow on Kubernetes)

https://tech.socarcorp.kr/data/2021/06/01/data-engineering-with-airflow.html

[socar-kyle]

안녕하세요! 블로그 글 재미있게 읽었습니다. 질문이 있어 코멘트 남겨봅니다.

1. 'Airflow는 항상 떠있지 않고 브랜치가 생성될때 작업을 할 수 있는 Airflow가 할당된다'라고 적어주신 airflow는 "Kubernetes 내부 DB를 사용했습니다" 와 연결되는것으로 이해했습니다. 그렇다면 제 생각에는 내부 DB에는 dag_run table에 기록이 저장되어 있지 않아 dag간의 실행 순서를 보장하는 `externalsensor` task 가 있는 DAG은 내부 DB에서 실제 동작한 task run 기록이 없어 개발 환경에서의 테스트가 어려울 것이라 생각드는데 개발 환경에서의 DAG 테스트는 어떻게 하고 계신지 노하우가 궁금합니다!

2. branch 마다 생성되는 airflow는 모두 다른 DB를 생성해서 연결되는것인지 하나의 내부 DB에 branch 마다 생성되는 airflow가 모두 연결되는 구조인지 궁금합니다.

This comment was made by Disqus. 2021-06-02 10:52:02 이병우

[socar-kyle]

안녕하세요.

1번 질문부터 하나씩 정리해보겠습니다.
- 말씀하신대로 내부 DB에는 dag_run table에 기록이 저장되어 있지 않습니다.
- `externalsensor` task 가 있는 DAG이 있는 경우, 의존하는 DAG을 수동으로 ON 해주고 트리거 걸어줘야 합니다. (에어플로우를 할당 받으면 모든 DAG은 기본적으로 OFF 상태입니다.)
- 기본적으로 자신이 작성한 DAG을 테스트하고 싶으면 할당받은 에어플로우에서 동작시키고 싶은 DAG을 ON을 걸어주고 결과를 확인해야 합니다. 각자의 작업 에어플로우마다 별도의 리소스, 환경을 가지고 있기 때문에 서로 영향 및 운영쪽에는 영향을 미치진 않습니다. 다만 결국 사람이 일일이 확인하기는 해야합니다.
- 솔직히 테스트하기 쉽지 않다고 생각합니다. 여전히 남은 숙제이긴 합니다. (이 때문에 가장 많이쓰고 있는 BigQueryOperator는 DBT같이 워크플로우로 분리하는 작업을 시도해보고 있습니다.)

2번 질문에 대한 답변입니다.
- branch 마다 생성되는 airflow는 모두 다른 DB를 생성한 뒤 연결합니다.
- 개발 환경에서는 각 에어플로우의 독립된 환경을 만드는 것이 가장 높은 고려사항이었습니다.

좋은 질문 주셔서 감사합니다 :)

This comment was made by Disqus. 2021-06-03 11:26:14 heumsi

[socar-kyle]

제 생각에는 개발 환경의 airflow 권한이 분리되어 있다면 개발환경의 airflow,내부 DB가 생성될때 운영환경의 dag_run table의 데이터를 export하여 내부 DB로 import 하여 말씀하신 "수동으로 ON 하는 부분의" 순서를 조금 더 자동화 할 수 있지 않을까라는 생각이 드는데 해당 부분도 어려울 것 이라 생각이 드네요. 말씀하신 테스트하기 쉽지 않다는 부분에 너무나 공감이 됩니다
질문에 친절히 답변 해주셔서 감사합니다 :)

This comment was made by Disqus. 2021-06-03 12:13:23 이병우

[socar-kyle]

좋은글 감사합니다 !
한가지 궁금한점이 있습니다 혹시 리소스를 따로 설정한 부분이 있으신가요 ?

This comment was made by Disqus. 2021-06-07 10:36:10 송우철

[socar-kyle]

리소스라면 쿠버네티스 리소스를 말씀하시는 걸까요?

This comment was made by Disqus. 2021-06-09 10:51:24 heumsi

[socar-kyle]

네 맞습니다. 혹시 operator별로도 리소스를 분배하시는지 궁금해서 물어봤습니다

This comment was made by Disqus. 2021-06-09 12:01:06 송우철

[socar-kyle]

에어플로우에서 제공해주는 오퍼레이터를 사용하는 경우, 하나의 pod_template_file.yaml를 사용하고 있습니다.
따라서 이 파일에 정의된 리소스대로 사용합니다.
경우에 따라서 Operator 에 resources를 추가적으로 넘겨주어서 따로 리소스 설정을 줄 수 있기는 하지만, 아직까지 필요한 경우는 없었습니다.
(아마 대부분의 DAG들이 I/O Bounded 테스크라 그렇게 헤비하게 리소스를 잡아먹지 않아서 그런듯 합니다.)

리소스를 따로 주는 부분은 주로 KubernetesPodOperator를 사용할 때 입니다.
이런 경우 일단 리소스를 대충 넉넉히 준 후, 그라파나를 통해 며칠 정도 리소스 사용량을 지켜본 뒤 적절히 줄여줍니다.

또한 팀에서 커스텀하게 개발한 Operator 역시 리소스를 따로 주고 있습니다.

This comment was made by Disqus. 2021-06-10 12:59:02 heumsi

[socar-kyle]

감사합니다 ㅎㅎ 충분히 도움이 되었습니다 !

This comment was made by Disqus. 2021-06-10 07:44:59 송우철

[socar-kyle]

좋은 글 감사합니다.
궁금한 점이 있어서 여쭤보고자 합니다.
질문의 요지는, 현재 블로그에 기술된 "Airflow를 각자에게 배포를 하여 테스팅 환경을 갖추는 것"이 얼마나 큰 장점을 갖는지 궁금합니다.

위의 본문에서 jupyter notebook으로 DAG 작성을 하셨다고 보았습니다.
Jupyter Notebook에 Git Extension으로 unit test는 jupyter에서, 이후 작성이 완료되고 제대로 DAG 포맷까지 잘 작성하였는지는 Argo workflow 또는 github actions등을 사용하여 CI 를 구축하는 것에 비해 현재 채택하신 개별 airflow 환경 - DAG 테스트 - git sync의 방식이 어떤 장점을 가질까요?

This comment was made by Disqus. 2022-03-10 11:36:21 Min Kang

[limkinam]

안녕하세요. 좋은 글 감사합니다.
궁금한 점이 있어서 여쭤봅니다.
airflow 운용시
dag : a - b - c - d - e
task a,b,c,d,e 의 경우
table A 에대해서
각 테스크가 A의 테이블 이름을 인자로 사용한다고 할때 a,b,c,d,e 에 어떻게 넘겨줄 수 있을까요?

실제로 사용해본 결과 dag 코드에서 table_name_A를 만들어서 넘겨주는 방식으로 설계했는데 실제로 table_name_A가 너무커서 (실제로는 api를 이용해 테이블 리스트를 가져오거나 날짜 등 연산작업 후 각 테스크에 들어갈 configure를 생성) table_name_A를 scheduler가 구문 분석하여 매번 실행시켜버리는 문제점을 가지고 있었습니다.

a,b,c 에서 매번 해당하는 작업에 필요한 인자를 만드는 작업을 하는건 중복작업을 너무 많이 하여 문제있는 방식이라고 생각하는데 혹시 어떤 식으로 처리 하셧는지 알 수 있을까요??
(현재 생각하고 있는 방식은 a-b-c-d-e 의 a 단계에서 각 task 에 필요한 인자를 만들어 xcom에 push 후 b,c,d,e에서 xcom에서 pull 하여 작업하는 방식인데 이 방법외에 다른 방법에 대한 조언을 구할 수 있을까요?)

[mystyle2006]

이번에 저희 팀도 MWAA를 도입해서 워크플로우 관리를 들어가는데 저희의 미래를 미리 당겨보는 기분이 들어 도움이 많이 됩니다.. 좋은 자료 감사해용 ^^

[Spidyweb-3588]

안녕하세요. 마찬가지로 구축하면서 궁금한점이 있습니다. airflow ui에서 확인해볼 수 있는 aws나 gcp connection은 어떤식으로 생성하셨나요? 클러스터 생성시에 secret 같은걸로 생성하셨는지, ui에서 직접 생성하셨는지 궁금합니다. ui로 생성한다면 ui에서 생성한 커넥션정보가 메타db에 남아잇으니 webserver pod가 바뀌어도 커넥션정보는 초기화되지않는 개념으로 보면 될까요?