<div id="readability-page-1" class="page"><div id="kakaoContent">

                <div>
                                            <p><a href="/m/category/%ED%94%84%EB%A1%9C%EA%B7%B8%EB%9E%98%EB%B0%8D/%EB%8D%B0%EC%9D%B4%ED%84%B0%20%EC%97%94%EC%A7%80%EB%8B%88%EC%96%B4%EB%A7%81">프로그래밍/데이터 엔지니어링</a></p><h2>Airflow 2 버전으로 Migration 한 후기</h2>
                    <p><cite>
                        <span>그랩 자유로운 오랑우탄</span>
                    </cite>
                </p></div>
                <p><time>2021. 8. 15. 18:04</time>
                    <span>
                        
                                                    <a href="#comment" onclick="Utils.smoothScroll(this);return false">
                                
                            </a>
                        
                                                    <span id="title_daumlike">
                                
                                <span id="mobile-reaction"></span>
                                
                            </span>
                                            </span>
                </p>

                
                <div>
                    
<figure data-origin-width="477" data-origin-height="131" data-filename="socar gif.gif" width="400" height="110" data-ke-mobilestyle="widthOrigin"><span data-url="https://blog.kakaocdn.net/dn/9tB8j/btrb7adZEXd/fRIsqY10STVRFpiO4mjqu0/img.gif" data-lightbox="lightbox" data-alt="migration하면서 화나서 귀요미 로고로 바꿔봤다"><img src="https://blog.kakaocdn.net/dn/9tB8j/btrb7adZEXd/fRIsqY10STVRFpiO4mjqu0/img.gif" data-origin-width="477" data-origin-height="131" data-filename="socar gif.gif" width="400" height="110" data-ke-mobilestyle="widthOrigin"></span><figcaption>migration하면서 화나서 귀요미 로고로 바꿔봤다</figcaption></figure>
<p data-ke-size="size16">데이터 엔지니어 포지션으로 회사에 들어간 지 3개월이 되었습니다. 취준을 하면서 데이터 파이프라인에 중요한 역할을 하는 Airflow를 처음 알게 되었고, execution_date와 start_date가 계속 헷갈렸었는데... 어느덧 회사에서 성공적으로 Airflow 1.10.14 버전에서 2.1.0 버전으로 migration을 완료하였습니다. 본 글에서는 Airflow 2로 migration의 레슨 런을 정리해봤습니다. 잘 정제된 글은 아닙니다만, 해당 주제로 조만간 회사 기술 블로그에서 더 상세하게 다루도록 하겠습니다.<span></span></p>
<h3 data-ke-size="size23"><span>&nbsp;</span></h3>
<h3 data-ke-size="size23"><span>Airflow v2</span></h3>
<ul data-ke-list-type="disc">
<li>Airflow란?
<ul data-ke-list-type="disc">
<li>Airflow는 Python으로 개발된 워크플로우 작성 + 스케줄링 관리 플랫폼입니다. 간단한 스케줄링 툴은 많지만, airflow의 매력은 태스크를 바탕으로 다양한 플로우 작성이 가능하고 실패한 작업들에 대해 BackFill을 잘 지원해주기 때문이라고 생각합니다.</li>
</ul>
</li>
<li>Airflow v2의 특징
<ul data-ke-list-type="disc">
<li>스케줄링 퍼포먼스 개선<br><figure data-origin-width="1280" data-origin-height="475" data-ke-mobilestyle="widthOrigin"><span data-url="https://blog.kakaocdn.net/dn/BUslM/btrci8SML12/ks0mbG2PAjhwwbtq9Sy6U1/img.png" data-lightbox="lightbox" data-alt="참고: 에어플로우 공식 문서&amp;nbsp;"><img src="https://blog.kakaocdn.net/dn/BUslM/btrci8SML12/ks0mbG2PAjhwwbtq9Sy6U1/img.png" data-origin-width="1280" data-origin-height="475" data-ke-mobilestyle="widthOrigin"></span><figcaption>참고: 에어플로우 공식 문서&nbsp;</figcaption></figure>
<ul data-ke-list-type="disc">
<li>Airflow 1버전 대의 가장 큰 문제는 Dag들이 늘어났을 때 Scheduler가 Dag run과 Task instance를 바로바로 실행시키지 못한다는 점입니다.</li>
<li>Airflow 2버전대에서는 DAG Serialization과 Fast-Follow를 도입하여 Scheduler의 반복적인 DAG 파싱 작업을 줄이고 Task Scheduling 과정이 개선되었습니다. <a href="https://www.astronomer.io/blog/airflow-2-scheduler">astronomer 블로그</a>에서 벤치마크 테스트를 했을 때 10배 이상의 성능 개선이 있었다고 합니다(확실히 연속된 Task instance 간의 Lag가 많이 줄긴 했습니다)</li>
<li>Scheduler HA를 지원해서 Scale Out이 가능하도록 지원합니다. Active-Active 클러스터 형태로 Scheduler의 HA를 지원합니다. (Meta DB에서는 SELECT ... FOR UPDATE로 row-level rock이 걸리게 됩니다. 그런데 테스트를 해보니 Scheduler들은 race 형식으로 스케줄링을 하게 되어 DB에 부담이 가지 않을까 하는 생각이 들었습니다)</li>
</ul>
</li>
<li>Web 서버 사용성 개선
<ul data-ke-list-type="disc">
<li>(Airflow 1.10 버전 대에도 있긴 하지만) Dag Serialization을 통해 더 빠르게 웹 UI에서 더 빠르게 DAG 정보를 불러올 수 있습니다.</li>
<li>Auto Refresh 기능이 추가되었습니다.</li>
<li>UI/UX가 모-던해졌습니다.</li>
</ul>
</li>
<li>Airflow Core Component에서 Provider 분리, Task Group, Smart Sensor, full rest API 도입 등등 꽤 많은 변화가 있습니다. 더 궁금하신 분들은 <a href="https://www.astronomer.io/blog/airflow-2-scheduler" target="_blank" rel="noopener">해당 글</a>을 읽어보시면 도움이 될 것 같습니다.</li>
</ul>
</li>
</ul>
<h3 data-ke-size="size23">migration 하는 과정</h3>
<ul data-ke-list-type="disc">
<li>Airflow 공식 문서에 <a href="https://airflow.apache.org/docs/apache-airflow/stable/upgrading-to-2.html#">2 버전 대로 마이그레이션 하는 가이드</a>가 있습니다. apache-airflow-upgrade-check 패키지를 설치하여 검사하고 Airflow db upgrade 명령어로 2버전 대에 맞게 업데이트하는 과정이죠.</li>
<li>하지만 메이저 버전이 올라갔는데 script로만 처리하기엔 운영 환경에서 불안정하다고 생각이 들었습니다 (통수를 몇 번 맞고 나니.. 애초에 안 믿었던 걸 수도) 그래서 저희 쪽에선 Airflow 2 버전을 위한 별도의 환경을 구축하고 천천히 migration을 했습니다.<br>
<ol data-ke-list-type="disc">
<li>Meta DB(Mysql), Log Storage(Google Cloud Storage) 등 Airflow 2를 위한 별도의 환경을 구축합니다.</li>
<li>airflow 2 버전의 DAG들이 정상적으로 파싱되는지, 로깅이 잘 되는지 등 버그들이 있는지 확인합니다.
<ul data-ke-list-type="disc">
<li>대부분 하위호환이 잘 되긴 했지만 몇 가지 부분에서는 DAG 코드 수정이 필요했습니다.
<ul data-ke-list-type="disc">
<li>Custom Operator를 생성할 때 kwargs에 custom 인자를 넣는 것이 금지되었습니다. 아래와 같이 별도의 파라미터를 넣어줘야 합니다.
<ul data-ke-list-type="disc">
<li>&nbsp;</li>
<li id="code_1629017551768" data-ke-language="python" data-ke-type="codeblock"><code>def **init**(self, prev_delta=0, *args, **kwargs):</code></li>
</ul>
</li>
<li>기존의 import 경로들은 Deprecated warning과 함께 하위호환이 되지만 안 되는 녀석(Sensor 같은 녀석)들도 몇 개 존재했습니다. 그럴 경우 Import Error를 뱉어 Dag 파싱 자체가 되지 않았습니다(만약 import 경로에 관해 하위 호환을 주고 싶다면 DAG Code 레벨에서 별도로 분기 처리를 해줘야 할 것 같다는 생각이 듭니다)
<ul data-ke-list-type="disc">
<li>
<pre id="code_1629017605979" data-ke-language="python" data-ke-type="codeblock"><code>(전) from airflow.sensors.external_task_sensor import ExternalTaskSensor
(후) from airflow.sensors.external_task import ExternalTaskSensor</code></pre>
</li>
</ul>
</li>
</ul>
</li>
</ul>
</li>
<li>기본적으로 특정 그룹 DAG들의 catchup을 False로 준 후, Airflow 1에서 OFF, Airflow 2에서 ON을 하면서 정상적으로 동작하는지 확인합니다. 2개 이상의 DAG RUN이 성공적으로 실행되었다면 무사히 잘 동작한다고 판단하였습니다.</li>
<li>정상적으로 migration이 완료한 후 1주일 정도 경과를 지켜보고 문제가 없다고 판단되면, Airflow 1 버전을 내립니다.</li>
</ol>
</li>
</ul>

<h3 data-ke-size="size23"><span>옮기면서 겪었던 문제점들</span></h3>
<ul data-ke-list-type="disc">
<li>Dag Serialization 과정이 포함된 Dag Parsing 과정이 생각보다 느립니다.
<ul data-ke-list-type="disc">
<li>저희 팀에서는 누구나 Airflow를 테스트해볼 수 있도록 개발 환경에서 격리된 Airlfow application을 띄워줍니다. 이때 Dag가 전부 파싱되는데 20분 이상이 걸리는 문제가 있습니다. 이에 .airflowignore 를 활용하여 특정 directory에 넣은 dag들만 파싱할 수 있도록 환경을 개선하였습니다. 이는 스케줄러의 CPU, Memory 자원 절감으로도 이어졌습니다.</li>
</ul>
</li>
<li>Scheduler의 작업이 CPU, Memory 성능을 많이 먹습니다. 초반에 Dag Serialization 작업이 들어가면서 (Dag 파싱과 스케줄링 프로세스가 분리되었고 더 많은 리소스를 먹지 않았나라는 추측을 해봅니다) 이를 최소화하기 위해서 대표적으로 몇 가지 configuration을 수정했습니다.
<ul data-ke-list-type="disc">
<li>airflow_scheduler_min_file_process_interval 높이기</li>
<li>airflow_scheduler_parsing_processes 낮추기</li>
<li>airflow_core_min_serialized_dag_update_interval 높이기</li>
</ul>
</li>
<li>자잘한 버그들이 계속 있기에 minor version 체크를 계속하고 코드 커스터마이징이 필요합니다.
<ul data-ke-list-type="disc">
<li>Log Line Break이 안 되는 이슈</li>
<li>Auto Refresh 이슈</li>
<li>등등</li>
</ul>
</li>
<li>Airflow Code를 Clone 한 후 Dockerfile을 이용해 Container build를 하는 방식으로 커스터마이징을 진행했는데, 버그 픽스를 위해 버전업을 하려고 할 때 Local Build가 정상적으로 동작하지 않습니다(고쳐줘!!😭)</li>
</ul>

<h3 data-ke-size="size23">느낀 점</h3>
<figure data-origin-width="582" data-origin-height="502" width="350" height="302" data-ke-mobilestyle="widthOrigin"><span data-url="https://blog.kakaocdn.net/dn/vaNbj/btrb8hjqjGt/GxG3LuC06idMF0NKsDKg70/img.png" data-lightbox="lightbox" data-alt="에어플로우야 다시 보지 말자~!"><img src="https://blog.kakaocdn.net/dn/vaNbj/btrb8hjqjGt/GxG3LuC06idMF0NKsDKg70/img.png" data-origin-width="582" data-origin-height="502" width="350" height="302" data-ke-mobilestyle="widthOrigin"></span><figcaption>에어플로우야 다시 보지 말자~!</figcaption></figure>
<ul data-ke-list-type="disc">
<li>기존 v1에서는 특정 시간 대에 Dag run이 수십 개가 동시에 실행될 때 Task Instance 스케줄링 Latency가 굉장히 길었습니다. 하지만 Airflow 2 넘어오면서 확실히 latency가 많이 줄었고 더 안정적으로 파이프라인 운영이 가능해졌습니다.</li>
<li>Kuberetes 환경에서 운영하면 Helm chart도 꽤 신경을 써야 했습니다. 어떤 차트를 쓰냐에 따라 kubernetes에서 failover가 잘 안될 수도 있었습니다. 저희 같은 경우<span>&nbsp;</span><a href="https://github.com/airflow-helm/charts/tree/main/charts/airflow">User Community Helm chart</a>를 사용하였으나, Airflow에서 공식으로<span>&nbsp;</span><a href="https://airflow.apache.org/docs/helm-chart/stable/index.html">Helm Chart</a>를 지원해준다고 하여 나중에 한 번 POC를 해볼 생각입니다.</li>
<li>뭔가 결과적으로 봤을 때 많은 코드 변화가 있는 것은 아닌데, stable 하게 팀 내에 도입하기까지 시간이 오래 걸렸습니다. 제가 airflow, kubernetes에 대한 배경지식과 히스토리가 부족한 상태에서 시작하다 보니 시행착오가 더 많았지만 Airflow의 복잡한 configuration이 또 한몫을 하는 것 같습니다(남 탓)</li>
</ul>

<p data-ke-size="size16">처음에 Kubernetes도 처음 써보고 Airflow를 다뤄본 경험도 적다 보니 정말 막막했습니다. 동료들이 도움을 주시지 않았다면 이곳에 작고 소중한 경험 공유마저 하지 못했을 것 같습니다. 본 주제에 대한 더 상세한 내용은 조만간 회사 기술 블로그에서 더 상세하게 다뤄보겠습니다. 감사합니다.</p>
                                                            
                    
											
					
                    
                    
                </div>

                                    
                
                
                                    
                
                


                

                                    
                
                
            </div></div>

그랩의 블로그

데이터 엔지니어 포지션으로 회사에 들어간 지 3개월이 되었습니다. 취준을 하면서 데이터 파이프라인에 중요한 역할을 하는 Airflow를 처음 알게 되었고, execution_date와 start_date가 계속 헷갈렸었는데... 어느덧 회사에서 성공적으로 Airflow 1.10.14 버전에서 2.1.0 버전으로 migration을 완료하였습니다. 본 글에서는 Airflow 2로 migration의 레슨 런을 정리해봤습니다. 잘 정제된 글은 아닙니다만, 해당 주제로 조만간 회사 기술 블로그에서 더 상세하게 다루도록 하겠습니다.
 
Airflow v2
Airflow란?
Airflow는 Python으로 개발된 워크플로우 작성 + 스케줄링 관리 플랫폼입니다. 간단한 스케줄링 툴은 많지만, airflow의 매력은 태스크를 바탕으로 다양한 플로우 작성이 가능하고 실패한 작업들에 대해 BackFill을 잘 지원해주기 때문이라고 생각합니다.
Airflow v2의 특징
스케줄링 퍼포먼스 개선
Airflow 1버전 대의 가장 큰 문제는 Dag들이 늘어났을 때 Scheduler가 Dag run과 Task instance를 바로바로 실행시키지 못한다는 점입니다.
Airflow 2버전대에서는 DAG Serialization과 Fast-Follow를 도입하여 Scheduler의 반복적인 DAG 파싱 작업을 줄이고 Task Scheduling 과정이 개선되었습니다. astronomer 블로그에서 벤치마크 테스트를 했을 때 10배 이상의 성능 개선이 있었다고 합니다(확실히 연속된 Task instance 간의 Lag가 많이 줄긴 했습니다)
Scheduler HA를 지원해서 Scale Out이 가능하도록 지원합니다. Active-Active 클러스터 형태로 Scheduler의 HA를 지원합니다. (Meta DB에서는 SELECT ... FOR UPDATE로 row-level rock이 걸리게 됩니다. 그런데 테스트를 해보니 Scheduler들은 race 형식으로 스케줄링을 하게 되어 DB에 부담이 가지 않을까 하는 생각이 들었습니다)
Web 서버 사용성 개선
(Airflow 1.10 버전 대에도 있긴 하지만) Dag Serialization을 통해 더 빠르게 웹 UI에서 더 빠르게 DAG 정보를 불러올 수 있습니다.
Auto Refresh 기능이 추가되었습니다.
UI/UX가 모-던해졌습니다.
Airflow Core Component에서 Provider 분리, Task Group, Smart Sensor, full rest API 도입 등등 꽤 많은 변화가 있습니다. 더 궁금하신 분들은 해당 글을 읽어보시면 도움이 될 것 같습니다.
migration 하는 과정
Airflow 공식 문서에 2 버전 대로 마이그레이션 하는 가이드가 있습니다. apache-airflow-upgrade-check 패키지를 설치하여 검사하고 Airflow db upgrade 명령어로 2버전 대에 맞게 업데이트하는 과정이죠.
하지만 메이저 버전이 올라갔는데 script로만 처리하기엔 운영 환경에서 불안정하다고 생각이 들었습니다 (통수를 몇 번 맞고 나니.. 애초에 안 믿었던 걸 수도) 그래서 저희 쪽에선 Airflow 2 버전을 위한 별도의 환경을 구축하고 천천히 migration을 했습니다.
Meta DB(Mysql), Log Storage(Google Cloud Storage) 등 Airflow 2를 위한 별도의 환경을 구축합니다.
airflow 2 버전의 DAG들이 정상적으로 파싱되는지, 로깅이 잘 되는지 등 버그들이 있는지 확인합니다.
대부분 하위호환이 잘 되긴 했지만 몇 가지 부분에서는 DAG 코드 수정이 필요했습니다.
Custom Operator를 생성할 때 kwargs에 custom 인자를 넣는 것이 금지되었습니다. 아래와 같이 별도의 파라미터를 넣어줘야 합니다.
 
def **init**(self, prev_delta=0, *args, **kwargs):
기존의 import 경로들은 Deprecated warning과 함께 하위호환이 되지만 안 되는 녀석(Sensor 같은 녀석)들도 몇 개 존재했습니다. 그럴 경우 Import Error를 뱉어 Dag 파싱 자체가 되지 않았습니다(만약 import 경로에 관해 하위 호환을 주고 싶다면 DAG Code 레벨에서 별도로 분기 처리를 해줘야 할 것 같다는 생각이 듭니다)
(전) from airflow.sensors.external_task_sensor import ExternalTaskSensor
(후) from airflow.sensors.external_task import ExternalTaskSensor
기본적으로 특정 그룹 DAG들의 catchup을 False로 준 후, Airflow 1에서 OFF, Airflow 2에서 ON을 하면서 정상적으로 동작하는지 확인합니다. 2개 이상의 DAG RUN이 성공적으로 실행되었다면 무사히 잘 동작한다고 판단하였습니다.
정상적으로 migration이 완료한 후 1주일 정도 경과를 지켜보고 문제가 없다고 판단되면, Airflow 1 버전을 내립니다.
 
옮기면서 겪었던 문제점들
Dag Serialization 과정이 포함된 Dag Parsing 과정이 생각보다 느립니다.
저희 팀에서는 누구나 Airflow를 테스트해볼 수 있도록 개발 환경에서 격리된 Airlfow application을 띄워줍니다. 이때 Dag가 전부 파싱되는데 20분 이상이 걸리는 문제가 있습니다. 이에 .airflowignore 를 활용하여 특정 directory에 넣은 dag들만 파싱할 수 있도록 환경을 개선하였습니다. 이는 스케줄러의 CPU, Memory 자원 절감으로도 이어졌습니다.
Scheduler의 작업이 CPU, Memory 성능을 많이 먹습니다. 초반에 Dag Serialization 작업이 들어가면서 (Dag 파싱과 스케줄링 프로세스가 분리되었고 더 많은 리소스를 먹지 않았나라는 추측을 해봅니다) 이를 최소화하기 위해서 대표적으로 몇 가지 configuration을 수정했습니다.
airflow_scheduler_min_file_process_interval 높이기
airflow_scheduler_parsing_processes 낮추기
airflow_core_min_serialized_dag_update_interval 높이기
자잘한 버그들이 계속 있기에 minor version 체크를 계속하고 코드 커스터마이징이 필요합니다.
Log Line Break이 안 되는 이슈
Auto Refresh 이슈
등등
Airflow Code를 Clone 한 후 Dockerfile을 이용해 Container build를 하는 방식으로 커스터마이징을 진행했는데, 버그 픽스를 위해 버전업을 하려고 할 때 Local Build가 정상적으로 동작하지 않습니다(고쳐줘!! )
 
느낀 점
기존 v1에서는 특정 시간 대에 Dag run이 수십 개가 동시에 실행될 때 Task Instance 스케줄링 Latency가 굉장히 길었습니다. 하지만 Airflow 2 넘어오면서 확실히 latency가 많이 줄었고 더 안정적으로 파이프라인 운영이 가능해졌습니다.
Kuberetes 환경에서 운영하면 Helm chart도 꽤 신경을 써야 했습니다. 어떤 차트를 쓰냐에 따라 kubernetes에서 failover가 잘 안될 수도 있었습니다. 저희 같은 경우 User Community Helm chart를 사용하였으나, Airflow에서 공식으로 Helm Chart를 지원해준다고 하여 나중에 한 번 POC를 해볼 생각입니다.
뭔가 결과적으로 봤을 때 많은 코드 변화가 있는 것은 아닌데, stable 하게 팀 내에 도입하기까지 시간이 오래 걸렸습니다. 제가 airflow, kubernetes에 대한 배경지식과 히스토리가 부족한 상태에서 시작하다 보니 시행착오가 더 많았지만 Airflow의 복잡한 configuration이 또 한몫을 하는 것 같습니다(남 탓)
 
처음에 Kubernetes도 처음 써보고 Airflow를 다뤄본 경험도 적다 보니 정말 막막했습니다. 동료들이 도움을 주시지 않았다면 이곳에 작고 소중한 경험 공유마저 하지 못했을 것 같습니다. 본 주제에 대한 더 상세한 내용은 조만간 회사 기술 블로그에서 더 상세하게 다뤄보겠습니다. 감사합니다.

Airflow 2 버전으로 Migration 한 후기

댓글