알루 코딩

이번에는 Xcom을 이용해서 email을 전송하는 실습을 해볼까 합니다. Pyython Operator의 결과값을 이용해서 Email 을 전송하는데, 중간에 Xcom을 사용해보려고 합니다. 먼저 (Python → Email 오퍼레이터 Xcom 전달)을 하기 위해서 Email 오퍼레이터는 어떤 파라미터에 Template를 쓸 수 있는지 알아봐야 합니다. 공식문서를 참고해보면, 'to', 'subject', 'html_content', 'files' 파라미터가 템플릿 문법을 적용할 수 있는 변수들입니다. 이 중에서 실습은 'subject'와 'html_content' 의 두 가지 필드에 템플릿 문법을 사용해보겠습니다. @task(task_id='find_dog_task') def dog_find(**kwar..

python 오퍼레이타에서 xcom을 사용하는 방법과 bash 오퍼레이터에서 xcom을 사용하는 방법을 배웠으니, 이제는 두 오퍼레이터를 혼합해서 Xcom 데이터를 구하는 방법을 알아보겠습니다. 1. Python → Bash 오퍼레이터 Xcom 전달 코드를 먼저 살펴보겠습니다. @task(task_id='python_push') def python_push_xcom(): result_dict = {'status':'MyLove','data':['Allu','Arsenal','Myself'],'options_cnt':100} return result_dict bash_pull = BashOperator( task_id = 'bash_pull', env = { 'STATUS':'{{ti.xcom_pull(t..

이번에는 Bash Operator에서 Xcom 사용하는 방법을 살펴보도록 하겠습니다. 1. Bash 오퍼레이터에서 Xcom 사용하기 먼저, 공식문서에서 Bash Operator에서 쓸 수 있는 템플릿 파라미터가 뭐가 있는지 알아야 합니다. 'env'와 'bash_command' 파라미터에서 템플릿 문법을 적용시킬 수 있는데, 이를 이용하여 push/pull 을 해봅시다.먼저 예시 코드를 보겠습니다. bash_push = BashOperator( task_id = 'bash_push', bash_command = "echo START && " "echo XCOM PUSHED " "{{ti.xcom_push(key='bash_pushed',value='first_bash_message') }} && " "e..

1. Xcom이란 무엇일까요? (=Cross Communication) Airflow DAG 안 Task 간 데이터 공유를 위해 사용되는 기술입니다. (ex) Task1의 수행 중 내용이나 결과를 Task2에서 사용 또는 입력으로 주고 싶은 경우가 있을 수 있겠죠? 주로 작은 규모의 데이터 공유를 위해 사용합니다. (Xcom 내용은 메타 DB의 xcom 테이블에 값이 저장됩니다.) 1GB 이상의 대용량 데이터 공유를 위해서는 외부 솔루션 연동이 필요합니다. (AWS S3, HDFS 등) 2. Python 오퍼레이터에서 Xcom 사용하기 크게 두 가지 방법으로 Xcom 사용이 가능한데요, 한 번 살펴보시죠. 1) **kwargs 에 존재하는 ti (task_instance) 객체 활용 예시 코드를 먼저 보..

BashOperator에서도 macro변수를 어떻게 쓸 수 있는지 알아본만큼, 이번 시간에는 PythonOperator에서 macro를 사용하는 방법을 알아보겠습니다. 1. 먼저 PythonOperator에서는 어떤 파라미터가 Template 변수를 지원할까요? 공식문서를 살펴보면, 'template_dict', 'op_args', 'op_kwargs' 가 템플릿을 지원합니다. 저번에는 op_kwargs 를 이용해서 template 변수를 써봤으므로, 이번에는 templates_dict 를 써서 macro변수를 써보겠습니다. Python 오퍼레이터를 사용해 macro변수를 써보는 코드를 작성해보겠습니다. @task(task_id = 'task_using_macros', templates_dict = {'..

Macro 변수는 Jinja 템플릿 내에서 날짜 연산을 가능하게끔 해주는 기능으로서 파이썬의 datetime이나 dateutil 같은 라이브러리를 이용해서 날짜연산을 할 수 있도록 지원을 해주고 있습니다. 1. 먼저 이런 Macro변수가 왜 필요한지 살펴봅시다. 아래상황을 가정해봅시다. DAG 스케줄은 매일 말일에 도는 스케줄인데(ex 0 0 L * *), BETWEEN 값을 전 월 마지막일부터 어제 날짜까지 주고 싶은데 어떻게 할까요? sql = f``` SELECT NAME, ADDRESS FROM TBL_REG WHERE REG_DATE BETWEEN ?? AND ?? ''' 와 같은 sql 이 있다고 합시다. 여기서 예를 들어, 배치일이 1월 31일이면, 12월 31일부터 1월 30일까지 배치일이..

저번에는 Bash Operator를 Jinja 템플릿을 사용했다면, 이번에는 Python Operator에서 사용해봅시다. 1. Python 오퍼레이터에서 with Template Python 오퍼레이터는 어떤 파라미터에 Template을 쓸 수 있을까요? 공식문서를 살펴보면 아래와 같습니다. python_callable op_kwargs op_args template_dict template_exts show_return_value_in_logs op_kwargs 와 op_args, template_dict 세 개의 파라미터가 template을 쓸 수 있습니다. 바로 한 번 실습을 해보도록 하겠습니다. 이 중에서, op_kwargs를 이용해서 jinja 템플릿을 써보도록 하겠습니다. (※ 파이썬 오퍼레..

이전 포스팅에서 bash operator를 이용해 템플릿 변수를 출력해봤는데, 치환된 값이 어떤 값인지 이해하기 위해서 먼저 Airflow에서의 날짜개념을 짚고 넘어가야 할 필요가 있습니다. 1. Airflow 날짜 Template 변수 이해 먼저, 데이터 추출 예시를 살펴봐봅시다. (ex) 등록 테이블 REG_DATAE NAME ADDRESS 2023-02-24 15:34:35 홍길동 Busan 2023-02-24 19:14:42 김태희 Seoul 2023-02-24 23:52:19 조인성 Daejeon Daily ETL 처리를 위한 조회 커리 (2023/02/25 0시 실행) 라고 가정합시다. (그렇게 되면, 24일에서 25일 사이의 데이터를 가져오는 셈이겠죠?) 쿼리를 작성해보면 아래와 같습니다. ..

Bash Opertor를 쓰면서 Jinja 템플릿을 어떻게 하면 적용할 수 있는지 살펴봅시다. 1. Bash 오퍼레이터 Bash 오퍼레이터는 어떤 파라미터에 Template를 쓸 수 있을까요? 공식문서를 살펴보면 아래와 같습니다. bash_command (str) (templated) env (dict[str,str] | None) append_env (bool) output_encoding (str) skip_exit_code (int) cwd (str | None) bash_command 와 env 두 개의 파라미터가 template을 쓸 수 있습니다. 바로 한 번 실습을 해보도록 하겠습니다. dags_bash_with_template.py 라는 파일을 dags폴더에 생성해 아래와 같이 코드를 짰습니..

에어플로우에서는 Jinja 템플릿을 활용하고 있기 때문에, 먼저 Jinja 템플릿에 대해 이해해보도록 하겠습니다. 1. Jinja 템플릿 문서(파일)에서 특정 양식으로 작성된 값을 런타임시 실제 값으로 치환해주는 처리 엔진입니다. 템플릿 엔진은 여러 솔루션이 존재하며 그 중 Jinja 템플릿은 파이썬 언어에서 사용하는 엔진입니다. ※ Jinja 라이브러리는 airflow를 설치할 때, 이미 설치가 됩니다. 예시 코드를 살펴봅시다. from jinja2 import Template template = Template('my name is {{name}}') new_template = template.render(name='allu') print(new_template)# my name is allu {{n..