[python] macOS에서 Apache Airflow 시작하기 (ETL 파이프라인 구축)

1.Airflow 초기 설정 

Apache airflow는 Linux 기반이지만, macOS에서도 터미널을 통해 쉽게 설치하고 실행할 수 있습니다.

 

1.1. 가상환경 생성(conda)

anaconda를 사용해서 python 3.10 버전의 가상환경을 생성 합니다.

# Anaconda를 사용하여 Python 3.10 환경을 생성
conda create -n airflow_lab python=3.10

# 가상환경 활성화
conda activate airflow_lab

1.2. Airflow 설치

Airflow 공식 문서에서는 pip 로 설치를 권장하기 때문에 Airflow와 MySQL Provider 패키지는 pip 로 설치 합니다.
mysqlclient는 conda로 설치해주었습니다.

# Airflow 설치
pip install apache-airflow

# MySQL Provider 설치
pip install apache-airflow-providers-mysql

# [중요] MySQL Client 설치 (이거 안 하면 에러 남)
conda install -c conda-forge mysqlclient

1.3. 환경 변수 설정

macOS의 시스템 보안 정책으로 인해 Airflow가 갑작스럽게 종료되는 현상이 발생할 수 있습니다. 이를 방지하기 위해 다음 환경변수를 설정합니다. 

 

⚠️ 주의: 이 설정은 현재 터미널에만 적용됩니다.

해당 명령어는 현재 열려 있는 터미널 창을 닫으면 초기화됩니다. Airflow를 실행할 때마다 해당 명령어를 입력해 주어야 에러가 발생하지 않습니다. 

# macOS 프록시 설정 (보안 정책 충돌 방지)
export no_proxy="*"

1.4. 서버 실행

이제 Airflow 서버를 실행합니다. 실행 후 터미널에 표시되는 웹 주소(일반적으로 http://localhost:8080)로 접속하면 Airflow 대시보드를 확인할 수 있습니다.

airflow standalone

---

2. Airflow 대시보드 접속하기

2.1.터미널에서 인증정보 확인하기

airflow standalone 명령어를 실행하면 여러 로그가 빠르게 출력됩니다.

로그가 나타나타가 멈추는 순간, 위쪽을 잘 살펴보면 다음과 같은 박스를 찾을 수 있습니다.

터미널에 출력된 Password 문자열을 복사해야 합니다.

Standalone mode created: ...
User: admin
Password: [여기에_적힌_복잡한_문자열]  <-- 이걸 복사하세요!

 

※ 주의: 만약 로그에 비밀번호가 안 보인다면?
Airflow 재실행 시에는 보안상 비밀번호가 로그에 뜨지 않습니다.
이런 경우에는 새 터미널 창을 열어 아래 명령어중 하나를 실행하세요.

# 비밀번호 확인 명령어 (둘 중 하나 실행)
cat ~/airflow/standalone_admin_password.txt

# 또는 (Airflow 버전에 따라)
cat ~/airflow/simple_auth_manager_passwords.json.generated

admin: "이 부분이 비밀번호 입니다."

2.2.웹 브라우저에 접속하기

웹 브라우저를(Chrome 등) 열고 주소창에 아래 주소를 입력합니다.

http://localhost:8080

 

2.3.로그인 하기

Airflow 로그인 화면이 표시되면 앞서 확인한 인증정보를 입력합니다.

- Username: admin 입력하시면 됩니다.

- Password: 위에서 복사한 비밀번호를 입력해줍니다.

로그인에 성공하면 Airflow 대시보드가 나타납니다.

---

3.실습용 데이터 생성하기 

Airflow와 연결하기 전에, Airflow가 가져올 실습용 데이터를 MySQL에 먼저 만들어 둬야 합니다.

 

3.1. MySQL 접속하기
새로운 터미널 창을 열고 (command + n) 터미널에서 MySQL에 관리자 권한으로 접속합니다.

(비밀번호가 있다면 입력 후 엔터, 비밀번호를 설정하신 적이 없다면 그냥 엔터!)

mysql -u root -p

3.2. 데이터 생성 SQL 실행 (복사+붙여넣기)
접속된 MySQL 화면(mysql>)에 아래 SQL 문을 통째로 복사해서 붙여넣으세요.

game_log_db라는 데이터베이스를 만들고, 그 안에 가상의 유저 로그 3개를 넣는 명령어입니다.

CREATE DATABASE IF NOT EXISTS game_log_db;
USE game_log_db;

CREATE TABLE IF NOT EXISTS access_log (
    user_id VARCHAR(50),
    event VARCHAR(50),
    created_at DATE
);

INSERT INTO access_log VALUES ('user_1', 'login', '2025-01-01');
INSERT INTO access_log VALUES ('user_2', 'logout', '2025-01-01');
INSERT INTO access_log VALUES ('user_3', 'level_up', '2025-01-01');

SELECT * FROM access_log;

3.3. 확인 및 종료
마지막에 SELECT 결과로 데이터 3줄이 잘 출력되었다면 성공입니다!
exit를 입력해 MySQL을 빠져나옵니다.

---

4.Airflow - MySQL 커넥션 설정 

Airflow에서 MySQL 데이터베이스에 접근하려면 먼저 연결 설정을 등록해야 합니다.

Airflow 웹 대시보드의 '관리자' 메뉴에서 커넥션을 추가합니다.

경로: 좌측 탭 > 관리자 > 커넥션들 > 커넥션 추가

 

4.1.기본 정보 입력

- 커넥션ID: game_mysql_conn

(Airflow 가 이 연결 설정을 부를 때 사용하는 고유한 식별자입니다. python 코드에서 이 이름으로 DB에 접속합니다.)

- 커넥션 유형: Mysql
(데이터베이스 유형을 지정합니다.)

 

4.2.데이터베이스 연결 정보 입력 

맥북 로컬 환경에서 에러 없이 연결하려면 아래 4가지는 꼭 채워야 합니다

 

1. Host (호스트): localhost

(주의: 127.0.0.1로 쓰면 맥북 보안 설정 때문에 연결이 거부될 수 있습니다.)

 

2. Login (로그인): root

(MySQL 설치 시 기본 관리자 계정입니다.)

 

3. Password (비밀번호): (본인 설정에 따라 다름)

(Mysqk울 설치할 때 비밀번호를 따로 설정 안 했다면 비워두세요.)

 

4. Schema:game_log_db

(앞서 생성한 데이터베이스 이름입니다.)

 

4.3.Mac OS 호환성 설정

(이걸 입력해야 TCP/IP 통신 오류를 우회하여 소켓으로 직통 연결됩니다.

경로는 터미널에서 mysql -u root -> status로 확인하세요.) 

{"socket": "/tmp/mysql.sock"}

 

왜 이 설정(Mac OS 호환성 설정)이 필요한가?

macOS에서 MySQL은 기본적으로 Unix Socket을 통해 통신합니다.

TCP/IP로 연결하려고 하면 ‘Connection Refused’ 에러가 발생할 수 있습니다.

이 설정은 Airflow에게 소켓 통신 방식을 사용하도록 지시합니다.

MySQL 소켓 경로 확인하기 터미널에서 다음 명령어를 실행하여 정확한 경로를 확인하세요.

Copymysql -u root
mysql> status

출력된 정보에서 'UNIX socket'으로 시작하는 라인을 찾으면, 소켓 파일의 경로를 확인할 수 있습니다 (보통 /tmp/mysql.sock).

만약 /var/run/mysqld/mysql.sock이라면, Extra 필드를 그에 맞게 수정하면 됩니다.

 

 

 

저장을 누르고 커넥션 목록 화면으로 돌아옵니다.

리스트에 game_mysql_conn이 추가된 것을 확인할 수 있습니다.

이제 Airflow가 MySQL에 접근할 수 있는 경로가 설정되었습니다.

이 'game_mysql_conn'이라는 연결 정보를 사용하여 다음 단계에서 Python DAG 코드를 작성할 예정입니다.

---

5.DAG 파일 생성(ETL 구현)

Airflow와 MySQL 연결 설정이 끝났으니, 이제 실제로 데이터를 추출(Extract)해서 파일로 저장(Load)하는 DAG를 작성해 봅시다.

 

5.1. 데이터를 저장할 폴더 만들기

CSV 파일을 저장할 폴더가 없으면 에러가 발생할 수 있습니다. 터미널에서 미리 폴더를 생성해 줍니다.

mkdir -p ~/airflow/data

5.2. DAG 파일 생성하기

Airflow의 DAG 폴더 (~/airflow/dags/)에 Python 파일을 생성합니다.

이 DAG는 다음 기능을 수행합니다.

 

- Extract: MySQL의 access_log 테이블에서 모든 데이터를 조회(SELECT)

- Load: 조회된 데이터를 CSV 파일로 로컬 컴퓨터 저장

 

다음 명령어를 터미널에 복사하여 실행하면 DAG 파일이 생성됩니다.

(터미널에 아래 내용을 그대로 복사+붙여넣기 하세요. 파일 생성과 내용 작성이 한 번에 됩니다!)

cat <<EOF > ~/airflow/dags/practice_connection.py
from airflow import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.operators.python import PythonOperator
from datetime import datetime
import csv

# ============================================
# 1단계: 데이터를 받아서 CSV 파일로 저장하는 함수 (Load)
# ============================================
def save_to_csv(**context):
    '''
    XCom(Airflow 우편함)에서 이전 Task가 조회한 데이터를 받아서
    CSV 파일로 저장합니다.
    '''
    # XCom에서 'read_db' Task의 결과 데이터를 꺼내옵니다
    data = context['ti'].xcom_pull(task_ids='read_db')
    
    # ⚠️ 중요: 아래 경로를 본인 사용자 이름에 맞게 수정하세요!
    # 예: /Users/[본인_사용자명]/airflow/data/game_log.csv
    file_path = '/Users/song/airflow/data/game_log.csv'
    
    # CSV 파일에 데이터를 쓰기 모드로 저장
    with open(file_path, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['user_id', 'event', 'date'])  # CSV 컬럼 제목
        for row in data:
            writer.writerow(row)  # 각 행의 데이터 입력
            
    print(f"✅ 파일 저장 완료: {file_path}")

# ============================================
# 2단계: DAG(작업 흐름) 정의
# ============================================
with DAG(
    dag_id='mysql_to_csv_v1',        # Airflow 웹 대시보드에 보일 DAG 이름
    start_date=datetime(2025, 1, 1), # DAG 시작 날짜
    schedule='@daily',               # 매일 UTC 0시(한국시간 9시)에 자동 실행
    catchup=False                    # 과거 날짜의 미실행 작업을 실행하지 않음
) as dag:

    # [Task 1] MySQL 데이터베이스에서 데이터 조회하기 (Extract)
    read_db = SQLExecuteQueryOperator(
        task_id='read_db',                  # Task의 고유한 이름
        conn_id='game_mysql_conn',          # 아까 웹에서 설정한 MySQL 커넥션 ID
        sql="SELECT * FROM access_log;"     # 실행할 SQL 쿼리
    )

    # [Task 2] 조회한 데이터를 파일로 변환하기 (Load)
    write_csv = PythonOperator(
        task_id='write_csv',                # Task의 고유한 이름
        python_callable=save_to_csv         # 실행할 Python 함수
    )

    # Task 의존성 설정: read_db를 먼저 실행 → write_csv를 나중에 실행
    read_db >> write_csv
EOF

 

---

6.실행 및 결과 확인 (Result)

코드를 저장했다면 이제 Airflow 웹 화면에서 실행해 볼 차례입니다.

 

DAG 확인 및 실행

1. Airflow 웹 화면(localhost:8080)에서 [DAGs] 메뉴를 클릭하고 새로고침(F5)을 합니다.

2. 약 30초~1분 정도 기다리면 목록에 mysql_to_csv_v1 이 나타납니다.
(만약 나오지 않는다면, Airflow 서버를 껐다 키신 후 진행해보세요)

3. 왼쪽의 토글 스위치(On/Off)를 켜서 파란색(Unpaused) 상태로 만듭니다.

4. 오른쪽의 [▶] (Trigger DAG) 버튼을 눌러 작업을 실행합니다.

5. 성공 여부 확인 (Web UI)

read_db와 write_csv 두 작업이 모두 진한 초록색(Success)으로 변하면 성공입니다!

(만약 빨간색(Failed)이 뜬다면, 클릭해서 [Log]를 확인해야 합니다.)

6. 최종 결과물 확인 (CSV 파일)

마지막으로 터미널에서 아래 명령어를 입력해 진짜 파일이 생성되었는지 확인합니다.

# 파일 내용 출력하기
cat ~/airflow/data/game_log.csv

아래와 같이 MySQL에 들어있던 데이터가 콤마(,)로 구분된 텍스트 형태로 출력된다면 ETL 파이프라인 구축 성공입니다!