Fake Onlineshop 프로젝트 (1) - 데이터 생성 및 적재

지난 포스팅 - Fake Onlineshop 프로젝트 (0) - 프로젝트 개요

 

🛒데이터 생성

데이터 파이프라인을 구축하기 위해서 가장 중요한 것은 데이터이다. 데이터를 구하는 것이 항상 문제였었는데, 이번에는 데이터를 직접 만들어서 사용하려고 한다. 이번 주제는 가상 쇼핑몰이기 때문에 쇼핑몰 데이터에서 가장 핵심이 되는 유저와 주문 데이터를 만들 것이다. 그 중 먼저 유저 데이터를 생성하는 작업을 해보도록 하자!

🙎‍♂️유저 데이터 생성

유저 데이터 생성에는 Faker 라이브러리를 사용하였다. 이 라이브러리는 한국어 정보도 추출할 수 있고, 무엇보다 유저의 간단한 프로필을 뽑을 수 있어서 데이터를 정의하는데 크게 고민할 필요가 없다는 점이 편하다. Faker 안에 simple_profile()을 이용하면 아주아주 간편하게 필요한 유저 프로필을 받을 수 있다.

 

{'username': 'kbag',
 'name': '강현숙',
 'sex': 'F',
 'address': '경상북도 청주시 청원구 역삼로',
 'mail': 'dohyeongang@hanmail.net',
 'birthdate': datetime.date(1978, 11, 13)}

 

하지만 여기에는 유저 번호가 없기 때문에 이를 따로 추가해주기만 하면 필요한 유저 정보 데이터는 모두 생성이 끝난다. (왜 유저 번호가 5981까지 있는지는 뒤에 주문 데이터 생성 부분에서 설명하도록 하겠다😀)

 

def get_users():
    fake = Faker('ko_KR')
    users = []

    cust_ids = [i for i in range(0, 5982)]
    for cust_id in cust_ids:
        simple_profile = {'userid': cust_id}
        simple_profile.update(fake.simple_profile())
        users.append(simple_profile)
    
    return users

🛒주문 데이터 생성

이제 주문 데이터를 생성할 차례이다. 사실 유저 데이터보다 더 까다롭기 때문에 어떤 식으로 생성할 지 고민이 좀 됐었다. 그런데 문득!!! 예전에 학교 수업에서 프로젝트를 할 때 받은 주문 데이터가 있다는 것이 생각났다! 고객번호부터 구매지점, 구매 상품, 구매 금액까지 다 나와있는 데이터라 여기에 바로 사용하면 좋을 것이라는 생각이 들었다.

 

모든 클라우드를 다 뒤져서 해당 데이터를 발견했다. 데이터를 데이터프레임 형태로 살펴보면 아래와 같다. 나름 잘 되어있는 데이터셋이라 이 프로젝트에서 사용하기에는 손색없는 것 같다. 그리고 실제 데이터이기 때문에 Fake로 만든 것보다 조금 더 당위성도 생기는 점도 장점이다.

 

 

이렇게 필요한 데이터는 모두 준비가 완료되었다!! 이제 해야할 작업은 준비된 데이터를 DB에 저장하는 것이다. DB에 데이터를 저장하러 한 번 가보자!!

💾DB 저장

저번 프로젝트에서는 미리 테이블을 만들어놓고 그 안에 데이터를 적재하는 방식이었는데, 이번에는 처음에 명령어 하나로 알아서 DB 구축부터 저장까지 되는 원터치(?) 기능으로 만들 계획이기 때문에 완전 기초부터 구축하는 코드를 작성하기로 하였다. 가장 먼저 DB를 만들어보도록 하자!

🧱스키마, 테이블 생성 및 데이터 추가

DB에 저장하기 위해 먼저 스키마와 테이블을 생성해야 한다. 온라인 쇼핑몰이라는 컨셉이기 때문에 먼저 store 스키마를 생성하고 기존에 만들어 놓은 데이터를 기준으로 하여 customers(유저)와 orders(주문) 테이블을 생성하였다.

 

DROP SCHEMA IF EXISTS store CASCADE;
CREATE SCHEMA store;
DROP TABLE IF EXISTS store.customers;
CREATE TABLE store.customers (
    userid      INT            not null,
    username    VARCHAR(20)    not null,
    name        VARCHAR(20)    not null,
    sex         VARCHAR(2)     not null,
    address     VARCHAR(50)    not null,
    mail        VARCHAR(30)    not null,
    birthdate   DATE           not null,
    
    PRIMARY KEY (userid)
);
DROP TABLE IF EXISTS store.orders;
CREATE TABLE store.orders(
    transactionid    INT         not null,
    custid           INT         not null,
    trandate         DATE        not null,
    store_nm         VARCHAR(15) not null,
    goods_id         INT         not null,
    gds_grp_nm       VARCHAR(30) not null,
    gds_grp_mclas_nm VARCHAR(30) not null,
    amount           INT         not null,
    
    PRIMARY KEY (transactionid),
    FOREIGN KEY (custid) REFERENCES store.customers(userid)
        ON DELETE CASCADE
);

 

기초 SQL 문법이라 그렇게 어렵지는 않았지만 처음에 스키마를 DROP할 때 아래에 구축된 테이블들이 모두 연관되어 있기 때문에 오류가 발생하는 문제가 있었다. 이 부분은 마지막에 CASCADE를 붙여 해결하였다. 나머지는 어렵지 않으므로 설명은 생략하도록 하겠다.😉

 

테이블을 생성했다면 데이터를 추가해야 한다. 해당 SQL도 어렵지 않지만 한 가지 주목할 점은 추가할 데이터를 format을 사용하여 placeholder로 만들었다는 것이다. 이에 대해서는 예전 중고책 프로젝트에서도 다뤘기 때문에 넘어가겠다.

 

user_placeholder = ', '.join(["%s" for _ in range(len(users[0]))])
INSERT INTO store.customers VALUES ({user_placeholder});

 

이렇게 CREATE와 INSERT 코드가 완성되었다. 이제 해당 SQL문을 이용하여 실제 명령을 보내어 DB를 구축하도록 해보자!

⌨️psycopg2를 이용하여 명령 보내기

이제 psycopg2를 이용하여 작성한 SQL문을 데이터베이스에 보내야 한다. 사실 이 과정은 저번 프로젝트에서도 했어서 과정이 어렵다고는 생각하지 않았다. 하지만 이번엔 테이블 생성부터 데이터 추가까지 모두 한 번에 해결하기 때문에 조금 더 깔끔한 방법이 필요하다고 생각했고 적절한 방법을 발견했다.

 

psycopg2를 사용한다면 DB에 연결하고(connect()) 닫는(close())이 동반되는데, 이를 한 번에 해결해줄 수 있는 파이썬의 with문을 이용하면 굉장히 깔끔하게 코드가 짜진다는 것을 알았다. 이를 이용하기 위해서는 DB에 연결하고 CRUD 할 수 있는 class를 하나 만들어 with문으로 객체를 생성하면 되겠다고 생각하였다.

 

아래는 위의 생각을 바탕으로 작성한 DatabaseConnect 클래스이다. 먼저 __init__ 메서드에서는 객체가 생성될 때 db에 연결할 connect() 객체를 하나 생성하도록 초기화시킨다. 연결에 필요한 파라미터들은 YAML 파일로 관리하도록 하였다.

 

참고로 초기화 메서드를 보면 self.engine과 self.conn 부분을 나누어 생성하도록 하였다. 전자는 sqlalchemy를, 후자는 psycopg2를 사용하는데 둘의 역할은 거의 동일하다고 볼 수 있다. 그렇다면 굳이 왜 따로 두었는지 의구심이 갈텐데 이 부분은 뒤에 가서 설명하도록 하겠다😎

 

이제부터가 with문에서 사용할 메서드를 정의하는 부분인데, __enter__와 __exit__가 그것이다. 이름에서도 알 수 있다시피 __enter__는 with구문이 실행되면, context manager에 의해서 실행되고 여기서 반환하는 값이 as 뒤의 변수에 들어가게 된다. 여기서는 DB에 연결한 conn의 cursor를 생성하고 객체를 리턴한다.

 

__exit__는 with문을 쓰는 가장 큰 이유이기도 한데, with문이 끝난 뒤 동작하는 메서드이다. 여기서는 DB에 연결할 때 사용한 conn과 cursor 객체를 종료하는데 사용한다. 

 

class DatabaseConnect:
    def __init__(self, db_conn):
        self.host = db_conn['host']
        self.username = db_conn['username']
        self.password = db_conn['pw']
        self.database = db_conn['database']
        self.port = db_conn['port']
        self.url = "postgresql://{}:{}@{}:{}/{}".format(self.username, self.password, self.host, self.port, self.database)
        self.engine = create_engine(self.url)
        self.conn = psycopg2.connect(self.url)
        
    def __enter__(self):
        if self.conn is not None:
            self.cursor = self.conn.cursor()
            return self
        else:
            raise IOError("Cannot access DB File")
            
    def __exit__(self, e_type, e_value, tb):
        self.cursor.close()
        self.conn.close()
        print("Closing Database")

 

기본적인 DB 연결을 위한 클래스 틀은 완성되었고 이제 본격적으로 DB에 CRUD하는 메서드를 만들어야 한다. 이 프로젝트에서는 CREATE와 UPDATE 정도만 필요하기 때문에 두 메서드만 구현을 하였다.

 

create 메서드의 경우 작성한 CREATE SQL을 바탕으로 실행되고, 그리 어렵지 않다. 주목해야할 부분은 insert 메서드이다. 여기에서 위에서 예고했던 create_engine을 사용한 이유가 나오게 된다. 

 

유저 데이터의 경우, 직접 만들어서 사용하기 때문에 psycopg2를 이용하면 되지만, 기존에 존재하는 데이터를 사용하는 주문 데이터는 데이터프레임으로 되어있기 때문에 데이터프레임의 to_sql 메서드를 사용하면 훨씬 편리하게 DB에 저장할 수 있다. 

 

하지만 이 메서드는 psycopg2로는 사용할 수 없고, sqlalchemy의 create_engine을 이용해야 하기 때문에 self.engine과 self.conn을 따로 생성한 것이다. 이를 이용하면 데이터 프레임을 명령어 한 줄로 아주 간편하게 DB에 적재할 수 있다.

 

    def create(self, sql):
        self.cursor.execute(sql)
        self.conn.commit()
        print("Create Complete!")
        
    def insert(self, users, df):
        user_placeholder = ', '.join(["%s" for _ in range(len(users[0]))])
        insert_sql = f"INSERT INTO store.customers VALUES ({user_placeholder});"
        for user in users:
            self.cursor.execute(insert_sql, list(user.values()))
            self.conn.commit()
        df.to_sql('orders', self.engine, schema = 'store', index = False, if_exists = 'replace')
        print("Insert Complete!")

 

이로써 DB에 적재하기 위한 클래스는 모두 완성되었다. 이를 이용하기 위해서는 먼저 만들어 둔 유저, 주문 데이터 생성 메서드(get_users, get_orders)로 데이터를 받는다. 그리고 with문을 이용해 DatabaseConnect의 객체를 생성하고, 작성한 메서드를 불러오면 된다. 참고로 나는 YAML 파일을 이용해서 파라미터를 관리하였기 때문에 YAML 파일을 불러오기 위한 with문도 작성하였다.

 

with open("data_generator.yaml") as f:
    db_conn = yaml.load(f, Loader=yaml.FullLoader)
    
users = get_users()
df = get_orders(db_conn['order_data_path'])

with DatabaseConnect(db_conn) as db:
    db.create(open("create.sql", "r").read())
    db.insert(users, df)

 

작성한 코드를 실행하고 데이터베이스를 확인해보면 아래와 같이 데이터가 잘 들어간 것을 볼 수 있다!

 

🚀앞으로

데이터를 생성하고 저장하는 작업까지 모두 마쳤다👏👏👏 사실 데이터 생성 작업은 이 프로젝트에서 큰 비중을 차지한 부분은 아니었지만 생각보다 많은 것을 배운 것 같다! 역시 배움의 길은 멀다...

 

이후에는 적재된 데이터를 웨어하우스에 적재하기 위한 적절한 형태로 변환하는 작업을 해야할 차례이다. ETL에서 T에 해당하는 부분으로 Spark를 많이 이용하지만 이번에는 dbt라는 새로운 도구를 이용해볼 생각이다. 처음 써보는 도구이기 때문에 다음 글에서는 간단하게 학습한 내용과 실제로 적용해보는 과정을 포스팅할 예정이다.