📦 Airflow ( CSV파일을 MySQL에 적재 )

이전에 작성했던 게시물에서는 data라는 폴더에 csv파일로 저장만 했다면
이제 저장된 csv파일을 Airflow에 connection한 MySQL에 적재를 해보겠습니다.

처음 mysql에 적재를 하기 위해서 로컬에 있는 mysql을 쓰려고 하다보니 간단하게 생각해 HOST에 localhost를 쓰면 될것이라고 생각을 했습니다.
하지만 그렇게 간단했다면 인생이 정말 재미가 없었을듯 합니다.
제가 부족하고 아직 많이 몰라서 이런일이 발생했는지는 모르겠습니다.

🖇️ Airflow에 MySQL connection하기

쉽게 과정을 설명하자면 airflow의 웹UI에서 Admin - connections에서 mysql을 추가하면 됩니다.

🖇️ 1. 자신이 쓰고있는 IP를 알아보기

ifconfig | grep inet

저는 mac을 사용하고 있기때문에 ipconfig쓰는것 대신에 ifconfig를 통해서 나온 ip중 inet을 통해 현재 쓰고있는 ip를 찾았습니다.

🖇️ 2. MySQL 환경 조성

저는 air라는 이름으로 계정을 새로 만들었습니다.

1. MySQL에 접속해서 유저를 만듭니다.

   CREATE USER air;
   

2. 비밀번호를 부여합니다.

   CREATE USER 'air'@'localhost' identified by '비밀번호';

3. 외부의 접근을 허용하기위해 host를 '%'로 바꿔줍니다.

   CREATE USER 'air'@'%' identified by '비밀번호';

4. 이렇게 만들어진 계정으로 들어가줍니다.
   mysql -u air -p

5. aladin이라는 스키마를 만들어주고 원래 크롤링했던 데이터들의 컬럼과 맞게 mytable이라는 테이블도 만들어 줍니다.

6. my.cnf에 bind-address를 외부에서 들어올수 있도록 0.0.0.0으로 바꿔주었습니다.
   하지만 이 방법은 보안상 매우매우 안좋은거 같습니다. 다른방법을 찾아봐야겠어요.

🖇️ 3. Airflow에 connection하기

Airflow 웹UI를 통해들어와 Admin - Connections에 들어오게 되면 여러가지 툴들을 커넥션 할수 있게 됩니다.
우리는 새로운 MySQL을 연결해야하기 때문에 왼쪽에 파란버튼으로 있는 + 버튼을 눌러서 추가해 줍시다.

이런 화면이 나오게 되는데 작성해야 할 것들을 하나씩 설명드리겠습니다.

• Connection Id : 커넥션의 이름과 같습니다. 추후에 스크립트에 이 ID를 사용해야합니다.
• Connection Type : 연결할 커넥션의 종류를 말합니다. (예 : mysql, aws 등등)
• Host : 사용하는 IP나 endpoint를 여기에 입력하면 됩니다.
• Schema : 데이터베이스의 이름을 말합니다.
• Login : 계정 이름입니다.
• Password : 계정의 비밀번호입니다.

위의 목록들을 다 작성하고 나면 커넥션이 완료가 됩니다.

제가 아직 부족해서 조언을 구하고 싶습니다.
💡 다른분들이 하는걸 보면 Host에 localhost를 작성 하시던데 저는 왜 안되는걸까요??

---

⚙️ DAG 작성

앞단의 데이터를 추출하는 코드는 앞의 게시글과 같습니다.

from airflow.hooks.base_hook import BaseHook
from sqlalchemy import create_engine
import pymysql

def read_csv_and_store_in_mysql():
    # CSV 파일을 읽어 DataFrame으로 변환합니다.
    today = str(date.today()).replace("-", "")
    df = pd.read_csv(f"/home/airflow/data/aladin{today}.csv")

    connection=BaseHook.get_connection('airsql') #airflow에서 connection한 mysql id
    database_username=connection.login
    database_password=connection.password
    database_ip = connection.host
    database_port = connection.port
    database_name = connection.schema
    
    database_connection = f"mysql+pymysql://{database_username}:{database_password}@{database_ip}:{database_port}/{database_name}"

    engine = create_engine(database_connection)

    # 데이터를 MySQL의 새 테이블에 저장합니다. 기존에 있는 테이블이라면, 데이터를 덮어쓰거나 추가할 수 있습니다.
    df.to_sql(con=engine, name='mytable', if_exists='append', index=False)
    
task_read_csv_and_store_in_mysql = PythonOperator(
	task_id='read_csv_and_store_in_mysql',
	python_callable=read_csv_and_store_in_mysql,
	dag=dag,
)

저는 alchemy를 통해서 MySQL서버와 연결하였습니다.
그리고 아까 connection했을때 보았던 것들이 많은데요.
BaseHook.get_connection('connectionID')를 통해서 connection한 내용을 불러왔습니다.

---

🗒️ 깨닳은것

csv파일로 저장해서 적재하는 이유

• 처음에는 추출한 데이터들을 하나의 딕셔너리에 담은후 다음 함수로 넘겨서 데이터프레임을 한 row씩 insert 하는 방식을 하였습니다.
  하지만 향후 데이터의 양이 많아지게 된다면 sql자체의 속도가 엄청나게 느려지면서 업무에 지장을 줄수 있다고 생각해 추출한 데이터를 csv파일에 저장하고 그것을 한번에 적재해야 속도가 느려지는것을 방지할수 있다는것을 배웠습니다.

---

🗒️ 향후계획

• 관계형 데이터 베이스가 아닌 csv파일 없이 NoSQL에 적재하기.
• NoSQL을 사용하려면 전처리를 해야하는지 안해야하는지 알아보기.