오늘 배운 것🤓

Connection

• 웹 UI를 통해 Connections를 추가할 때 써야하는 항목
  • Conn Id : 원하는 이름
  • Conn Type : DB 종류 (Postgresql, MySQL, Redshift, ...)
  • Host : DB 접속 주소
  • Schema : Schema 혹은 DB
  • Login : 접속 ID
  • Password : 접속 PW
• python에 적용하는 방법

# Redshift는 PostgresSQL을 사용하므로 postgres hook으로 접속한다.
from airflow.providers.postgres.hook.postgres import PostgresHook

hook = PostgresHook(postgres_conn_id='Connections에서 설정한 Conn id')
conn = hook.get_conn()
conn.autocommit = True  # default : False

task는 어느정도 분리해야하지?

• task 하나에 기능을 많이 넣으면(즉, task가 적으면) 모듈화가 되지 않아 에러를 찾고 수정하는데 오래 걸릴 수 있으며 실패 시 재실행에 시간이 오래 걸린다.
• task를 너무 잘게 쪼개면(즉, task가 많으면) task를 개별적으로 스케줄러가 스케줄링을 해야하므로 스케줄러에 부하가 간다.
• 따라서, task의 수를 너무 늘리지는 않되, 재실행이라는 이슈가 발생했을 때 재실행의 시간을 줄일 수 있는 방안을 고민하여 task를 분리해야 한다. (...예?)

airflow.cfg(환경설정파일) 중요 key

• dags_folder : DAGs 폴더가 지정되는 위치
• dag_dir_list_interval : DAGs 폴더에 새로운 DAG를 만들었을 때 Airflow 시스템에서 적용하는 시간(단위 : 초)
• [auth_backends] : Airflow를 API형태로 외부에서 조작하고 싶을 때 조작하는 섹션
• Variable의 값이 암호화되는 키워드 :
  • secret
  • password
  • passwd
  • authorization
  • api_key
  • apikey
  • access_token
• 환경 설정 파일을 수정했을 때 적용/반영하는 명령어
  $ sudo systemctl restart airflow-webserver
$ sudo systemctl restart airflow-scheduler
  도커의 경우 재실행
• fernet_key : MetaDB내용을 암호하고 싶을 때 지정하는 key

Incremental Update로 구현하는 순서

• Full-Refresh와는 다르게 기존에 있는 데이터에 새로운 데이터를 집어넣는 방식

1. 실제 테이블의 내용을 갖는 임시 테이블 생성
   CREATE TEMP TABLE tmp_table AS SELECT ~~~
2. 임시 테이블에 새로운 데이터 적재
3. 원본 테이블을 DROP & CREATE
4. 임시 테이블의 중복을 제거 후 INSERT
   INSERT INTO origin_table SELECT DISTINCT * FROM tmp_table

• CREATE TEMP TABLE은 말그대로 임시 테이블을 만드는 명령어이므로 사용 후 자동으로 삭제해주기 때문에 직접 DROP을 안해줘도 된다.
• 1~4모두 transaction으로 감싸도 되고 3~4 과정만 감싸도 된다.

cmd 이용 시 고려할 점 (docker 이용시)

• Airflow에서 Python 모듈을 추가하고 싶다면 scheduler container에서 설치하면 된다.
• DAG는 /opt/airflow/dags 위치에 있다.
• root 유저로 컨테이너에 접속하고 싶다면
  $ docker exec --user root -it 컨테이너id sh

느낀 점😊

차근차근 하나씩 직접해보니 너무 재밌다!!! 진도는 늦었지만 이렇게 하나씩 해보는 과정이 너무 즐거워서 시간이 훅훅 지나가버린다. 그래도 너무 늦으면 안되니까 시간을 고려하면서 진행해야겠다.