[데이터 파이프라인] 3. 데이터 수집 실습

이번 실습에서는 데이터 수집 중에서 로컬 데이터베이스에서 csv 파일을 추출 후 s3 버킷에 업로드 하는 것을 실습한다. 알아야 할 SDK는 boto3이다. boto3는 파이썬 HTTP 프레임워크로, AWS 서비스를 다루기 위해 만들어졌다. HTTP 요청과 응답을 사용하기 때문에 AWS 서비스만 쓰이는 것은 아니고 로컬의 여러 서비스 접속에도 활용할 수 있다.

 

오늘의 할 일

• pymysql을 이용한 로컬 MySQL 접속
• MySQL Orders 데이터를 csv 파일로 추출 후 로컬에 저장
• 로컬에 저장한 csv 파일을 boto3을 이용해 AWS s3 버킷에 저장

 

로컬 MySQL 접속

모든 실습은 ubuntu에서 진행한다. python 스크립트를 이용해 MySQL 데이터베이스에 접속하는 것이 목적이다. 참고로 ubuntu는 본 운영체제 위에서 돌아가는 가상 환경이기에 본 운영체제에 설치된 MySQL과 헷갈리면 안된다. 필자는 이것 때문에 몇 시간을 날렸다.

 

[mysql_config]
hostname = localhost
port = ****
username = ****
password = *****
database = ****_data_pipeline

 

.config 파일을 활용해 python 스크립트에 설정 변수를 쉽게 사용하자. 위는 ubuntu에서 돌아가는 MySQL 설정 변수 내용이다.

 

# local mysql database 접속
parser = configparser.ConfigParser()
parser.read("pipeline.conf")
hostname = parser.get("mysql_config", "hostname")
port = parser.get("mysql_config", "port")
username = parser.get("mysql_config", "username")
dbname = parser.get("mysql_config", "database")
password = parser.get("mysql_config", "password")

conn = pymysql.connect(
    host=hostname,
    user=username,
    password=password,
    db=dbname,
    port=int(port)
)

if conn is None:
    print("Error connecting to the MySQL database")
else:
    print("MySQL connection Established!")

 

위는 config 파일에서 가져온 설정 변수로 pymysql SDK를 이용해 mysql에 접속하는 conn 객체를 만드는 python 코드다. conn 객체로 로컬의 mysql 데이터베이스 객체에 접속해 데이터를 활용할 수 있다.

 

try-except 구문을 활용해 예외상황에 대응할 수 있지만 간단하게 if-else로 구현했다. try-except 구문은 다음 단계에서부터 활용된다.

 

 

MySQL 데이터를 csv 파일로 추출

# local mysql database 접속 후 쿼리 실행
try:
    m_query = "SELECT * FROM Orders;"
    local_filename = "order_extract.csv"

    m_cursor = conn.cursor()
    m_cursor.execute(m_query)
    results = m_cursor.fetchall()

    # 결과값 csv 파일로 정리
    with open(local_filename, 'w') as fp:
        csv_w = csv.writer(fp, delimiter='|')
        csv_w.writerows(results)

    fp.close()
    m_cursor.close()
    conn.close()
except Exception as e:
    print(f"MySQL query failed : {e}")

 

전 단원에서 생성한 conn 객체를 이용해 SQL 쿼리를 실행하고 넘어온 데이터들을 취합해 csv 파일로 추출하는 과정이다. 파일 포인터를 이용해 파일을 열고 데이터들을 저장한다. try-except 구문을 활용하여 예외상황을 처리했다.

 

 

스크립트를 활용해 csv 파일을 AWS s3 버킷으로 전송

로컬에 있는 csv 파일을 AWS s3 파일로 전송하는 과정이다. AWS에 접속하기 위해서는 key가 필요하다. key들은 AWS s3 버킷을 만들 때 확인할 수 있으며 한 번 보면 못보기 때문에 파일에 저장하는 것을 추천한다. (github에 올리면 안된다!)

 

[aws_boto_credentials]
access_key = **********
secret_key = ***********
bucket_name = *-data-pipeline
account_id = *******

 

이렇게 설정변수들을 구성하면,

# AWS s3 Bucket에 해당 데이터 전송
try:
    parser = configparser.ConfigParser()
    parser.read("pipeline.conf")
    access_key = parser.get("aws_boto_credentials", "access_key")
    secret_key = parser.get("aws_boto_credentials", "secret_key")
    bucket_name = parser.get("aws_boto_credentials", "bucket_name")

    s3 = boto3.client('s3', aws_access_key_id = access_key, aws_secret_access_key=secret_key)

    s3_file = local_filename
    s3.upload_file(local_filename, bucket_name, s3_file)
except Exception as e:
    print(f"AWS Bucket connection failed {e}")

 

s3 파일에 데이터를 전송하기 위한 코드를 작성한다. 이 과정에서 boto3 SDK가 활용된다. AWS 서비스를 다루기 위해 고안되었기 때문에 s3에 파일을 업로드하는 .upload_file 같은 메소드들을 제공한다. try-except 구문을 활용해 예외상황에 대비한다.

 

 

결과와 느낀점

 

csv 파일이 s3 버킷에 올라간 것을 확인할 수 있다.

 

처음으로 AWS 서비스를 사용해봤는데, 코드 따라 s3 버킷에 결과물이 올라가니 신기하다.