예니의 코딩일기

실습 실행 과정 1. airflow-setup Github repo clone git clone https://github.com/learndataeng/learn-airflow.git 2. airflow-setup/yml 다운로드 cd learn-airflow curl -LfO "https://airflow.apache.org/docs/apache-airflow/2.5.1/docker-compose.yaml" 3. docker-compose.yaml version x-airflow-common services postgres redis airflow-webserver airflow-scheduler airflow-worker airflow-trigger airflow-init volumes postgr..

강의 Backfill "Backfill"은 실패한 데이터 파이프라인 혹은 이미 추출한 데이터의 오류로 인한 데이터 파이프라인 재실행을 말합니다. Full Refresh : 단순 재실행 Incremental Update : 복잡 → Airflow Backfill 용이 날짜별 Backfill 결과 및 성공 여부 기록하여 ETL 인자로 제공합니다. 시스템이 지정해준 날짜 사용하면 Backfill을 구현할 수 있습니다. Airflow Backfill 구현 방식 ETL별 실행 날짜 및 결과 메타데이터 데이터베이스에 기록 모든 DAG 실행에 execution_date 지정 데이터 갱신하는 코드 작성 Backfill 관련 Airflow 변수 start_date : DAG가 처음 추출해야하는 데이터의 날짜 및 시간 e..

강의 CountryInfo 예제 Full Refresh UTC 기준 매주 토요일 오전 6시 30분 실행 import requests @task def extract_transform(): response = requests.get('https://restcountries.com/v3/all') countries = response.json() records = [] for country in countries: name = country['name']['common'] population = country['population'] area = country['area'] records.append([name, population, area]) return records with DAG( dag_id = ..

강의 Yahoo Finance API 예제 Full Refresh 구현 Yahoo Finance API 호출 애플 주식 정보 수집 (지난 30일) Redshift에 테이블 및 레코드 적재 트랜잭션 형태 구성 from airflow import DAG from airflow.decorators import task from airflow.providers.postgres.hooks.postgres import PostgresHook from datetime import datetime from pandas import Timestamp import yfinance as yf import pandas as pd import logging def get_Redshift_connection(autocommit=..

강의 DAG 파라미터 "DAG parameters"는 Task parameters와 다르게 DAG 객체를 만들 때 지정합니다. max_active_runs max_active_tasks catchup DAG 폴더 주의 사항 Airflow의 dags 폴더 주기적으로 스캔 → DAG 모듈의 모든 파일 메인 함수 실행 → 개발 중인 테스트 코드 실행 가능 Connetions와 Variables Connetions 일부 연결 관련 정보 저장 : hostname, port number, access credential Postgres 연결 혹은 Redshift 연결 정보 저장 Variables API 키 혹은 일부 구성 정보 저장 값 암호화 가능 : access 혹은 secret 사용 Xcom "Xcom"은 태스..

강의 Airflow 코드 기본 구조 DAG 대표하는 객체 생성 : DAG 이름, 실행주기, 실행날짜, 오너 DAG 구성 태스크 생성 : 오퍼레이터 선택. 태스크 ID 부여 및 작업 세부사항 지정 태스크 실행 순서 결정 DAG Trigger 방법 Airflow Web UI cmd airflow dags list airflow tasks list dag_v1 airflow tasks test dag_v1 ls 2020-08-09 airflow dags test dag_v1 2019-12-08 airflow dags backfill dag_v1 -s 2019-01-01 -e 2019-12-31 Bash Operator 사용 예시 from airflow import DAG from airflow.operator..