XCOM Xcom은 DAG 내의 task 사이에서 데이터를 전달하기 위해서 사용되는데, Xcom은 DAG내에서만 공유할 수 있는 변수라는 점이다. Xcom을 이용해 데이터를 전달하는 경우 DataFrame이나 많은 양의 데이터를 전달하는 것은 지원하지 않으며, 소량의 데이터만 전달하는 것을 권장한다. ti = task_instance 로 축약 가능 XCOM 사용 방법 PythonOperator return 값을 이용한 Xcom 사용 push-pull을 이용한 Xcom 사용 Jinja templates을 이용한 Xcom 사용 def _player_data_write(**context): headers = {'Authorization' : '='} player_url = "" data = requests.g..

Airflow 태스크 의존성 기본적인 태스크 download_launches >> get_pictures >> notify 병렬 태스크 [start, download_launches] >> get_pictures >> notify

BashOperator Bash shell 스크립트를 실행하는 오퍼레이터. 리눅스 명령어, 프로그램 실행이 가능함 # 예시 import airflow from airflow import DAG from airflow.operators.bash import BashOperator dag = DAG( dag_id="listing_4_01", start_date=airflow.utils.dates.days_ago(3), schedule_interval="@hourly", ) get_data = BashOperator( task_id="get_data", # bash 명령어 입력 bash_command=( "curl -o /tmp/wikipageviews.gz " "https://dumps.wikimedia.o..

DAG 스케줄링 cron Crontab.guru - The cron schedule expression editor crontab.guru dag = DAG( dag_id = "FIFA_Online4", description = "FIFA Online 4 Match Detail Analysis", start_date = datetime(2022,11,12, tzinfo = kst), end_date = datetime(2022,11,30, tzinfo = kst), dagrun_timeout=timedelta(minutes=300), schedule_interval = '0 2,7,14,20 * * *' ) 백필 ( Backfill ) 파라미터에 포함된 일시를 순서대로 바꿔가면서 일정 기간의 플로우를 연..

가상환경 설정 $ conda create --name airflow python==3.8 $ conda activate airflow Airflow 설치 $ export AIRFLOW_HOME=~/airflow $ AIRFLOW_VERSION=2.2.3 $ PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)" $ CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt" $ pip install "apache-airflow==${AIRFLOW_VERSI..

데이터파이프라인 소개 - 태스크를 병렬로 실행 -> 가용 컴퓨팅 리소스를 더 효율적으로 활용! - 파이프라인을 작은 점진적인 태스크로 명확하게 분리 할 수 있음. - 실패한 태스크만 재실행 하면됨 -> 효율적 ※ 모놀리식 : 파이프 라인의 중간 태스크가 실패하면 전체 스크립트를 재실행 해야함 -> 비효율적 Airflow DAG를 파이썬 코드로 정의할 때 장점 - 프로그래밍 접근 방식이 DAG를 구성하는데 많은 유연성을 제공할 수 있음. ※ DAG ( Directed Acyclic Graph, 방향성 비순환 그래프 ) Airflow 주요 구성요소 ▶ Airflow 스케줄러 DAG를 분석하고 현재 시점의 스케줄이 지난경우 Airflow 워커에 DAG의 태스크를 예약함. ■ Airflow 스케줄러 작업 단계 ..