Data Engineering

[Airflow] airflow db init

~/Desktop/git/airflow-pgt/airflow_celery   main  docker exec -it airflow-webserver bash airflow@airflow-webserver:/opt/airflow$ airflow db init /home/airflow/.local/lib/python3.8/site-packages/airflow/configuration.py:545: DeprecationWarning: The sql_alchemy_conn option in [core] has been moved to the sql_alchemy_conn option in [database] - the old setting has been used, but please update you..

[Airflow] context, kwargs 에 들어있는 값 확인하기

from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime import pprint dag = DAG ( dag_id = '4_python_operator_context', start_date = datetime(2022, 12, 3), schedule_interval = '* * * * *', catchup = False, tags = ['server_local', 'detail_test'], description = 'Python Operator Sample', default_args = {'owner': 'ParkGyeongTae'}) def func_p..

[Airflow] context를 이용한 xcom에 key, value 남기는 방법

case 1) from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime dag = DAG ( dag_id = '3_python_operator_xcom', start_date = datetime(2022, 12, 3), schedule_interval = '* * * * *', catchup = False, tags = ['server_local', 'detail_test'], description = 'Python Operator Sample', default_args = {'owner': 'ParkGyeongTae'}) def func_xcom_push_..

[Airflow] Xcom에 데이터 남기지 않는 방법

기존 DAG - Xcom에 데이터를 task마다 남긴다.. from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime dag = DAG ( dag_id = '2_python_operator', start_date = datetime(2022, 12, 3), schedule_interval = '* * * * *', catchup = False, tags = ['server_local', 'detail_test'], description = 'Python Operator Sample', default_args = {'owner': 'ParkGyeongTae'})..

[Airflow] PythonOperator에서 return값은 로그에 남는다.

case 1) from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime dag = DAG ( dag_id = '1_python_operator', start_date = datetime(2022, 12, 3), schedule_interval = '* * * * *', catchup = False, tags = ['test'], description = 'Python Operator Sample', default_args = {'owner': 'ParkGyeongTae'}) def print_1(): return '111' print_11 = PythonOp..

[Airflow] 각 컨테이너별로 생성되는 로그를 확인해보자

flower - /opt/airflow/logs - 아무것도 없음 init - /opt/airflow/logs - 아무것도 없음 scheduler - /opt/airflow/logs webserver - /opt/airflow/logs - 아무것도 없음 worker - /opt/airflow/logs

[Airflow] postgres MetaDB에 사용하지 않는 데이터가 쌓이는 현상

지금까지 확인한 상황은 다음과 같다. 1) 1_python_operator 라는 dag를 만들면 일어나는 일 - 웹서버에서 dag 표시가 됨 - postgres db에서 dag 테이블에 데이터가 생김 - postgres db에서 dag_code 테이블에 데이터가 생김 추가확인) 웹서버의 dag와 postgres db에서 dag 테이블의 데이터 생성이 일치함 2) 1_python_operator 라는 dag를 실행하면 일어나는 일 - task 실행됨 - dag_run 에 데이터 생김 - task_instance 에 실행한 task 데이터 생김 - flower 에 실행한 task 생김 3) 1_python_operator 라는 dag를 웹서버에서 삭제하면 일어나는 일 - 웹서버에서 dag 삭제됨 - post..

[Airflow] MetaDB에서 task 실행기록 확인하는 방법

1_python_operator 라는 dag를 2번 실행시켰고, task는 6개가 있다. select * from task_instance

[Airflow] DAG 추가시 MetaDB 변경사항 확인

기존 dag 는 1_python_operator 이고, 추가된 dag 는 2_python_operator 이다. 메타DB를 확인해보자 select * from dag select * from dag_code

[Airflow] 에어플로우 실행시 최초 Meta DB 테이블 확인