[TIL] Airflowdbt Day 1~5 ⭐

heering·2023년 6월 19일

데브코스 데이터엔지니어링

목록 보기

36/39

✔ 6/19

docker-compose.yaml

# 이렇게 설정해두면 웹 UI에는 안 보임. 하지만 웹 UI에서 세팅하는 거랑 똑같은 효과
environment:
    AIRFLOW_VAR_DATA_DIR: ~ # AIRFLOW_VAR로 시작하면 Airflow의 환경변수
    AIRFLOW_CONN_BLAH_BLAH: ~ # AIRFLOW_CONN으로 시작하면 Airflow 웹 UI에 세팅된 connection이랑 똑같음, 그 뒤의 string이 connection의 이름이 됨.

.airflowignore

Airflow의 DAG 스캔 패턴: dags_folder가 가리키는 폴더를 서브폴더 포함 모두 스캔해서 DAG 모듈이 포함된 모든 파이썬 스크립트를 실행함 → 가끔 사고로 이어짐
.airflowignore는 Airflow가 의도적으로 무시해야 하는 디렉토리/파일 지정
예시 (정규표현식)
- .airflowignore 안의 내용이 아래와 같다면
  - project_a → project_a_dag_1.py, TESTING_project_a.py, project_a/dag_1.py 이와 같은 파일 무시
  - tenant_[\d] → tenant_1.py 이와 같은 파일 무시

✔ 6/20

Airflow API 활성화

airflow.cfg의 api section 내용

[api]
auth_backend = airflow.api.auth.backend.basic_auth

docker-compose.yaml
AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session 여기에서 처음 언더바 2개 다음의 글자(API)는 section을 의미, 그 다음 언더바 2개 다음 글자(AUTH_BACKENDS)는 section 밑의 key를 의미.
is_active와 is_paused 의미
curl -X GET --user "airflow:airflow" http://localhost:8080/api/v1/dags 이와 같이 GET 요청을 했을 때 dags에 관한 정보들이 뜸. 여기서 is_active는 dags folder에 존재하는지 여부이고, is_paused는 현재 멈춰 있는지 (활성화된 dag인지 아닌지) 확인할 수 있는 값.

✔ 6/21

Airflow API - variables

환경변수로 지정된 건 return하지 않음. 예를 들어 DATA_DIR이라는 환경 변수 존재 확인하려면 $ docker exec -it {컨테이너_ID} airflow variables get DATA_DIR -> 결과 출력됨.

Jinja Template

bash_command (str), env (dict[str, str])는 Jinja Template 지원. 여기 참고해서, 파라미터 설명에 (templated)라고 되어있으면 Jinja Template 지원한다는 의미

Sensor

특정 조건이 충족될 때까지 대기하는 Operator
Airflow의 내장 Sensor들
- FileSensor: 지정된 위치에 파일이 생길 때까지 대기
- HttpSensor: HTTP 요청을 수행하고 지정된 응답이 대기
- SqlSensor: SQL DB에서 특정 조건을 충족할 때까지 대기
- TimeSensor: 특정 시간에 도달할 때까지 워크플로우를 일시 중지
- ExternalTaskSensor: 다른 Airflow DAG의 특정 작업 완료를 대기, 실수하기 쉬우므로 딱히 쓰지 않는 걸 추천하셨다.
기본 모드가 poke고, worker 하나를 붙잡고 거기서 계속 체크하고, 기다렸다가 또 체크하고... → worker 하나가 낭비되는 특징

DataGrip 스키마 안 보일 때

여기 참고했다. 저번에 팀 프로젝트 할 때도 안 보여서 어쩌다보니 해결했는데.. 실습할 때도 안 보이는 스키마가 있었다. 새로 뭘 하기 전에 앞으로는 미리 확인 해놓아야지

Airflow 메타데이터 DB 내용 확인

$ docker exec -it {CONTAINER_ID} sh
$ psql -h postgres
$ \dt
하고 싶은 쿼리 날리기

✔ 6/22 ~ 6/23

DBT - Sources

기본적으로 처음 입력되는 ETL 테이블을 대상으로 함
- 별칭(alias) 제공
- 최신 레코드 체크 기능 제공

Airflow dags 실행 안되는 문제 해결

터미널에서 airflow dags test로 하면 DAG 실행이 잘 되지만, localhost:8080에 들어가서 dag를 trigger하면 실행이 안되는 문제가 발생했다.

airflow-worker에서는 PermissionError: [Errno 1] Operation not permitted: ~~~가 떴다.
airflow-scheduler에는 Executor reports task instance <TaskInstance: {DAG_ID}.extract scheduled__2023-06-21T00:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally? 가 떴다.
그것도 모든 DAG가... 나한테 왜 이래. airflow-worker에서 알려준 경로에 log 파일이 존재해야 하는데 쓰기 권한이 있는데도 존재하지도 않았다. DAG는 trigger한 이후로 계속 running 상태 또는 다 fail.. 그래서 나름대로 해결해보려고 삽질을 일주일 내내 했다. 😫
원인?
일단 예상되는 원인은 airflow 2.5.1 자체에 버그가 있다는 것. 구글링 하다가, 간신히 나랑 똑같은 오류가 발생했다는 글을 찾았다. 나는 airflow 2.5.1 버전을 docker-compose해서 쓰고 있었는데 글쓴이의 버전도 나랑 똑같다. bug를 고쳐서 커밋하셨길래, 봤더니 try-except 구문으로 처리된 코드가 추가되어 있었다.

그래서 당장 내 환경에서 해당 부분을 확인해봤다. 경로는 docker container에 접속해서 여기로 이동했다. → /home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py

음.. 나에겐 그런 부분이 존재하지 않아 오류가 발생했던 것 같다. 물론 나도 확실히 모름 ㅎ
근데 여기서 내가 고치는 게 말도 안될 것 같아서 그냥 airflow 2.6.1 버전 docker-compose.yaml파일 새로 받고 다시 했다. 2.6.1 버전 확인해보니 해당 부분이 try-except로 처리되어 있었다. DAG 실행도 다시 잘 됐다. 👍

dbt Version 업그레이드 하기

pip3로 install하면 dbt 버전이 1.0.0 보다 한참 낮은 버전으로 설치되는 문제가 발생했다. 얘는 seeds 폴더를 못 알아먹는다,, 실습해야 하는데 마음에 안 든다. dbt가 1.0.0 버전 이후로는 pip3 install upgrade 조차 지원을 안한다고 한다. 그럼 어떻게 해결했느냐?

GitHub에서 가장 최근 버전인 dbt-core-1.5.2.tar.gz를 다운로드 받자.
저장하고 해당 파일 있는 곳에서 압축 풀자. $ tar -xvf dbt-core-1.5.2.tar.gz
안으로 들어가자. $ cd dbt-core-1.5.2
설치하자. $ python setup.py install
버전 확인하자. $ dbt --version
그럼 이제 postgres 또는 redshift 플러그인을 설치할 차례. 각자 하고 싶은 거 깔자. $ pip3 install dbt-redshift 그러면 core 버전에 맞게 plugin도 높은 버전으로 깔린다. 아래처럼!