'에어플로우 airflow' 카테고리의 글 목록

[Airflow] 컴포넌트, MWAA

2024.05.15

Airflow 컴포넌트Airflow는 복잡한 워크플로우를 스케줄링하고 관리하기 위한 오픈소스 플랫폼임.데이터 엔지니어 및 개발자들은 작업의 의존성을 명확히 정의하고, 반복적인 작업을 자동화할 수 있음.Airflow는 여러 컴포넌트로 구성되어 있으며, 각각의 역할이 있음 Airflow 각 컴포넌트 역할1. Web ServerFlask를 기반으로 한 웹 어플리케이션으로 사용자 인터페이스를 제공함.웹 서버를 통해 사용자는 작업 스케줄, 실행 상태 확인, 로그 보기 등의 작업을 수행할 수 있음. 2. Scheduler워크플로우의 스케줄링을 담당함.이 컴포넌트는 DAGs(방향성 비순환 그래프)를 주기적으로 폴링하며, 실행해야 할 새로운 작업 인스턴스를 확인하고 실행함. 3. Executor실제 작업을 실행하는 역..

에어플로우 airflow

[Airflow] 에어플로우 DAG란?

2023.04.02

DAG - Directed Acyclic Graph - 순환하지 않는 그래프 - 대그라고 부른다. - 순차적으로 작업이 실행된다. - 순환 실행을 하지 않는다.

에어플로우 airflow

[Airflow] 데이터 파이프라인이란?

2023.01.09

1. 언제, 어디에서, 어떻게, 왜 데이터를 수집할 것인가에 대한 고민 필요 2. 데이터 파이프라인 구축시 수동작업 제거 필요 3. 데이터가 흐르도록 만들어야 함 4. 데이터 파이프라인 구축은 추출, 변경, 결합, 검증, 적재 과정을 자동화하는 것 5. 여러 데이터 스트림을 한번에 처리가 가능해야 함 6. ETL은 추출, 변환, 적재의 줄임말 7. 데이터 파이프라인은 ETL을 포함하는 광범위한 말

에어플로우 airflow

[Airflow] 에어플로우의 로그파일이 존재하지 않을 때

2022.12.10

비정상 airflow-worker *** Log file does not exist: /opt/airflow/logs/dag_id=4_python_operator_context/run_id=scheduled__2022-12-10T11:22:00+00:00/task_id=print_kwargs/attempt=1.log *** Fetching from: http://airflow-worker:8793/log/dag_id=4_python_operator_context/run_id=scheduled__2022-12-10T11:22:00+00:00/task_id=print_kwargs/attempt=1.log *** !!!! Please make sure that all your Airflow components..

에어플로우 airflow

[Airflow] 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 your config. 에러 해결

2022.12.06

/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 your config. 에러 해결하는 방법 환경변수 변경 AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow 를 다음처럼 바꾼다 AIRFLOW__DATABASE__S..

에어플로우 airflow

[Airflow] airflow db reset을 하면 scheduler가 죽는다..

2022.12.06

~/Desktop/git/airflow-pgt/airflow_celery   main  docker exec -it airflow-webserver bash airflow@airflow-webserver:/opt/airflow$ airflow db reset /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 yo..

에어플로우 airflow

[Airflow] airflow db init

2022.12.06

~/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

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

2022.12.06

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

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

2022.12.06

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

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

2022.12.06

기존 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'})..

티스토리툴바