[final project] airflow를 활용한 wordcloud 생성(1): airflow 설치 및 DAG 순서도 작성

airflow 실습하기

로컬에서 설치하기

1. wsl에서 airflow 설치
   pip3 install apache-airflow

2. 터미널에 airflow라고 입력하면 관련 명령어가 잘 나오는지 확인.

3. cd ~에서 airflow 폴더 확인, cd airflow로 접속 후 ls

4. mkdir dags & airflow db init 후 ls
   

5. airflow 관리자 계정 생성
   airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com

6. airflow webserver와 scheduler 실행
   airflow webserver & airflow scheduler
   

7. airflow 로컬 웹페이지 접속
   localhost:8080 또는 127.0.0.1:8080으로 접속 가능, 앞서 생성한 admin 계정으로 접속(아이디 admin, 비밀번호 admin)
   

---

docker에서 airflow 컨테이너 띄우기

프로젝트 실전에서는 docker에서 airflow를 띄워 gcp db랑 연결하면 좋을 것 같아 로컬이 아닌 docker 컨테이너에서 airflow를 실행하는걸로 변경 !

1. 터미널에서 docker 실행
   sudo service docker start

2. docker에서 airflow 컨테이너 생성 및 실행
   docker run -it -p 8080:8080 -v /home/**username**/workspace/dags/:/opt/airflow/dags/ -v /etc/localtime:/etc/localtime:ro --entrypoint=/bin/bash --name airflow --privileged apache/airflow:2.6.1

위의 명령어를 사용했을 때, address already in use 오류 발생 시

• 앞서 로컬에서 설치한 airflow가 이미 8080 포트를 사용하고 있어 발생한 오류
• ps -ef | grep airflow: 현재 실행중인 airflow 관련 프로세스 확인
• 해당 프로세스의 PID를 확인하고 kill PID라는 명령어를 통해 프로세스 종료
• 해당 명령어를 한번 더 입력하면 name이 겹친다는 오류가 뜰 수 있는데, 이는 이전에 만들어둔 airflow 컨테이너를 삭제 후 다시 명령어를 입력하면 해결

3. airflow 입력창이 뜨면 airflow db init

4. airflow 관리자 계정 생성
   airflow users create --use rname admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com

5. airflow webserver & airflow scheduler 차례대로 실행
   

6. airflow 웹페이지 접속
   localhost:8080 또는 127.0.0.1:8080으로 접속 가능, 앞서 생성한 admin 계정으로 접속(아이디 admin, 비밀번호 admin)
   

7. 비어있는 /home/**username**/workspace/dags/ 폴더에 예시용 dag 파일을 하나 넣고, airflow webserver에서 제대로 실행되는지 확인
   

   

dags 폴더에 접근권한이 없다고 나올 경우

• (권한 부여)/home/workspace/에 들어가서 sudo chown **username** dags 입력
• (권한 확인) ls -l | grep dags
  

• 이후 dags가 잘 실행되는 것을 볼 수 있다.
  

---

Airflow Pipeline 만들기

MySQL 설정

1. 이전에 만들어놓은 db의 사용자 목록 확인
   

• Host중 root 라는 계정은 어떤 client에서든 접근이 가능하게 되어있다(%)

  만약 % 계정이 없을 경우

• (계정 생성)create user [username]@'%' identified by [password];
• (권한 부여)grant all privileges on [db명][table명] to [username]@'%';
  • 모든 DB의 모든 Table을 대상으로 할 경우 .
• (변경 반영)flush privileges
• (계정 접속)mysql -u [username] -p

MySQL 연결

1. 

2. library 설치

DAG 만들기

Operator 종류 알기

• Action Operator : 간단한 연산 수행 오퍼레이터, airflow.operators 모듈 아래에 존재. 실습에서 사용할 대부분의 오퍼레이터는 여기에 속한다.
• Transfer Operator : 데이터를 옮기는 오퍼레이터, <출발>To<도착>Operator 꼴.
• Sensor : 태스크를 언제 실행시킬 트리거(이벤트)를 기다리는 특별한 타입의 오퍼레이터 (예를 들어 어떤 폴더에 데이터가 쌓여지기를 기다린다든지, 요청에 대한 응답이 확인되기를 기다린다든지).

아래 DAG에서 사용하는 오퍼레이터 종류들

• PythonOperator : 파이썬 코드를 돌리는 작업을 할 때 사용하는 기계
• BashOperator : bash 명령어를 실행시키는 작업을 할 때 사용하는 기계
• MysqlOperator : MYSQL DB 사용과 관련된 작업을 할 때 사용하는 기계

DAG 작성

내가 하고 싶은 airflow 실습 코드의 순서도
1. 배틀그라운드 공식 카페 데이터 크롤링
2. 긴단한 전처리 및 DB 적재
3. 설정한 기간 데이터 가져와 합치기
4. 자연어 처리기 활용 단어 추출
5. 워드클라우드 생성 및 이미지 저장
(데이터를 가져와서 웹화면에 airflow로 띄우는 작업은 장고에서 할 예정)

task	method	operator
데이터 크롤링	배틀그라운드 공식 카페 데이터 크롤링	PythonOperator
데이터 적재	긴단한 전처리 및 DB 적재	pythonOperator
데이터 추출	설정한 기간 데이터 가져와 합치기	pythonOperator
단어 추출	자연어 처리기 활용 단어 추출	PythonOperator
워드클라우드	워드클라우드 생성 및 이미지 저장	PythonOperator

흠 .. 이게 맞나 (dag 파일 만들면서 나중에 수정)

대략적인 DAG TASK 순서도

• task1 함수 : 데이터 크롤링 전처리 후 gcp에 있는 db에 연결하고 db에 저장(1+2)
• task2 함수 : db에 저장된 article(subject, content, comment) 활용 기간만큼 불러와서 합치기(3)
• task3 함수 : 자연어 처리 및 워드클라우드 생성&저장(4+5)

task1 >> task2 >> task3

• task4 : 장고의 templates폴더로 보내는 작업(백엔드)

이 포스팅에 참고 및 활용한 문서들 !

• https://velog.io/@clueless_coder/Airflow-엄청-자세한-튜토리얼-왕초심자용
• https://velog.io/@moon_happy/Airflow-CSV파일을-MySQL에-적재