1. 코호트 분석이란? Cohort Analysis

코호트(동질 집단)란 무엇일까? 특정 기간동안 공통된 특성이나 경험을 공유하는 사용자 집단을 의미한다. 코로나 발생 초기에, 병원에서 코호트 격리를 한다는 기사가 많이 나왔는데 거기서 사용되는 의미와 동일하다.

코호트 분석(동질 집단 분석)은 이러한 유사 성질을 갖는 코호트를 만들어, 시간에 따른 각 코호트의 행동과 여러 지표들을 분석할 때 사용하는 기법이다.

여러 출처로부터 다양한 정의를 내리자면,

• 코호트 분석은 동종 집단이 나타내는 시간적 변모 양태를 분석하여 예측하고자 하는 연구이다.
• 코호트 분석은 사용자를 기간에 따라 그룹으로 분류하여 그룹의 행동과 유지율을 분석할 때 활용하는 기법이다.
• 코호트 분석은 시간 경과에 따라 유저 그룹을 추적하는 방법이다.

여기서 중요한 점은, 특정 기간 경험을 공유한다는 점과, 유사 속성을 갖는다라는 점이다.

코호트 테이블

이제 실제 코호트 분석에 쓰이는 테이블을 살펴보자. 출처는 GA에 있는 샘플 차트이고, 조건은 아래와 같다.

• 코호트 유형은 ‘획득 날짜’이다. 즉, 최초로 사이트에 접속한 날짜를 기준으로 코호트를 만들겠다는 의미이다.
• 코호트의 크기는 ‘일별’이다. 일 단위로 코호트를 분리하겠다는 의미이다.
• 측정 항목은 사용자 유지율이고, 7일간의 데이터를 살펴본다.

위 이미지에서 행(날짜 데이터)이 코호트이다. 간단히 해석해보자.

• 4월 10일, 1714명의 사용자가 최초로 접속했다. (사실 정말로 최초는 아닐테지만, 코호트의 시작점이라 그렇다).
• 4월 10일에 방문한 고객 중, 다음 날(+1일 열) 3.68%의 고객만이 재방문했다(유지되었다).
• 4월 10일에 방문한 고객 중, 6일 뒤(+6일 열), 0.88%의 고객만이 재방문했다.
• 4월 14일, 2149명의 사용자가 최초로 접속했다. (신규 유저)
• 4월 14일에 방문한 고객 중, 다음 날(+1일 열) 3.82%의 고객만이 재방문했다.

그럼 4월 14일에 방문한 총 고객은 몇 명일까? 2149명일까? 정답은 ‘그렇지 않다’이다. 코호트에 표기된 숫자는 새롭게 획득된 사용자 수를 의미한다. 즉, 4월 14일에 새롭게 유입된 고객이 2149명이라는 의미이다. 코호트 테이블만을 보고 4월 14일의 전체 방문자는 알 수 없다. 다만, 코호트 내의 값으로 어느정도 계산은 할 수 있다. 4월 14일의 방문자는 코호트 내의 아래 방문자들로 구성된다.

• 4월 14일의 신규 유입
• 4월 13일의 +1일 유저들 (재방문)
• 4월 12일의 +2일 유저들 (재방문)
• 4월 11일의 +3일 유저들 (재방문)
• 4월 10일의 +4일 유저들 (재방문)
• 그리고 코호트 기간을 벗어나는, 이전 방문자들 중 재방문자들

코호트 테이블의 구현

여기서 짚고 넘어가야 할 부분이 있다. 코호트 분석에서 집계 방식은 크게 두 가지가 있다.

1. 코호트의 유입 날짜를 기준으로 하는 것
2. 코호트의 유입 날짜와 직전 기간을 기준으로 하는 것 (Rolling retention)

예를 들어, 사용자 A가 4월 10일의 코호트에 속하는데, 4월 11일과 4월 13일에 방문했다고 해보자. 1번 방식으로는 4월 11일, 4월 13일이 모두 집계되지만 2번 방식에서는 4월 11일이 마지막 집계가 된다. 왜냐하면 4월 12일에는 방문을 하지 않았기 때문에, 직전 기간이 12일이 되는 13일도 집계가 되지 않는 것임.

이에 대한 자세한 내용을 찾아보려 했는데, 구체적으로 다룬 내용을 찾지 못했다. 나중에 찾게 되면 추가하겠음. GA는 1번 방식을 택하고 있다.

2. 코호트 분석의 의미

코호트 분석이 왜 중요한걸까? 개인적으로는 행동 분석(Behavioral Analytics)을 기반으로 하는 분석 기법들이 대부분 동일한 목적을 갖는다고 생각하는데, 데이터를 전체로 보는 것이 아닌 부분으로 나누어 보아야 행동 분석의 의미가 있기 때문이다.

• 이벤트를 기반으로 데이터를 나누게 되면, 퍼널 분석이나 AARRR과 같이 고객의 각 행동(이벤트)을 단계별로 나누어 각 단계를 면밀히 관찰할 수 있다.
• 사용자를 기반으로 데이터를 나누게 되면, 일반적인 세그먼트나 세그먼트의 일종인 코호트 분석, RFM 분석처럼 유사한 고객별로 면밀히 관찰할 수 있다.

이벤트를 중심으로 살펴보든, 사용자를 중심으로 살펴보든 유사한 특성을 지닌 단계(그룹)별로 나누어 보는 게 데이터 분석이 훨씬 용이하기 때문이다. 더 나아가, 사용자를 기반으로 타겟을 쪼개고 그 타겟별로 이벤트 기반 분석을 할 수 있다. 즉, 세그먼트별로 퍼널이 어떻게 다른지도 비교해 볼 수 있고 더 세분화된 분석이 가능하다. 이러한 세분화된 분석은 맞춤 마케팅을 가능하게 해준다.

좀 더 직접적인 의미를 찾자면, 코호트 분석은 고객의 유지율(그리고 이탈률)을 분석하는 데 탁월한 지표이다. 사용자가 감소하는 시기를 포착하여 개선방안을 제시할 수 있기 때문이다. 예를 들어, 사용자가 급격히 이탈하는 지점이 발생하면 그 시기에 대한 조사를 통해 문제가 있는 부분을 개선할 수 있고, 푸시 알림 등을 통해 이탈한 고객을 다시 유입시킬 수도 있다.

3. 코호트 분석 해보기

이제 실제 데이터셋을 통해 코호트 테이블을 그려보고, 간단한 분석을 진행해보자.

1) 데이터 소개

캐글에서 데이터를 가져왔고, 여기서 다운로드 받을 수 있다. ‘2019-Oct’ 부터 ‘2020-Feb’ 까지 5개 파일 모두 다운받으면 된다.

이 과정이 귀찮은 사람을 위해 이미 정제한 데이터도 올려두겠다. 여기를 눌러서 다운로드하기

데이터는 화장품을 다루는 이커머스가 출처이며, ‘상품 클릭 view’, ‘장바구니 담기 cart’, ‘장바구니 제외 remove_from_cart’, ‘구매 purchase’ 네 가지 이벤트에 대한 정보를 담고 있다. 아쉬운 점은 단순히 상품 클릭이 아닌, 사이트에 최초 접속한 레코드가 없다는 점인데 그냥 감안하고 진행하도록 하겠다.

2) 데이터 정제

여기에는 2019년 10월부터 2020년 2월까지 총 5개월의 데이터가 담겨있다. 하지만 개별 데이터가 너무 크기 때문에, 각 월별로 고유한 1000명의 데이터만 뽑아서 샘플링한 데이터를 쓴다.

더 구체적으로, 고유한 1000명에 더해 이전 기간의 코호트를 유지하기 위해 앞선 기간의 유저는 그대로 유지하는 방식을 택한다. 즉,

• 파일별 1000명을 샘플링하여, 해당 1000명의 데이터를 모두 가져온다.
• 이전 기간의 사용자를 기록해서, 이후 파일에서도 동일한 사용자가 나오면 포함한다.

import pandas as pd
import random

whole_target_pool = set()

for file in ['2019-Dec', '2019-Nov', '2019-Oct', '2020-Feb', '2020-Jan']:
    target = pd.read_csv('{}.csv'.format(file))
    sample_targets = random.sample(list(target.user_id.unique()), 1000)

    whole_target_pool.update(sample_targets)

    # 1000명의 샘플에 있거나, 이미 뽑은 앞선 목록에 있으면 타겟으로 넣는다.
    new_target = target[target.user_id.isin(sample_targets) | target.user_id.isin(whole_target_pool)]

    print(len(new_target.user_id.unique()), new_target.shape)
    new_target.to_csv('{}-1000.csv'.format(file), index=False)

1000 (9418, 9)
1141 (16677, 9)
1241 (17031, 9)
1270 (21948, 9)
1482 (27129, 9)

이제 5개의 파일을 하나로 합쳐서, 최종 파일을 만든다. 앞서 말한대로 이미 정제된 파일은 위에 링크를 달아두었다.

whole_df = pd.DataFrame()

for file in ['2019-Dec', '2019-Nov', '2019-Oct', '2020-Feb', '2020-Jan']:
    target = pd.read_csv('{}-1000.csv'.format(file))
    whole_df = pd.concat([whole_df, target])

print(whole_df.shape)
whole_df.head(3)

(92203, 9)

whole_df.to_csv("ecommerce_cosmetics_sampled.csv", index=False)

3) SQLite에 데이터 쌓기

다른 방식으로 진행해도 되지만, 일반적으로 데이터가 DB에 들어있다고 가정하고 SQL을 통해 코호트 테이블의 기초 데이터를 만들고자 한다.

여기서는 데이터베이스로 SQLite을 사용한다. SQLite은 이름 그대로 정말 라이트하게 사용할 수 있는 SQL DB 엔진이다. 별도의 설정도 필요 없고, 그냥 설치하고 곧바로 파이썬 소스로 데이터를 넣어주면 된다. 나는 맥OS를 쓰는데, 이미 설치가 되어 있어 생략했다. 여기서 최신 버전을 다운받자.

관계형 DB에서 가장 먼저 할 일은, DB 서버에 연결하고, 개별 DB에 접속하는 일이다. SQLite은 별도로 서버 개념이 없기 때문에, 실제로 코드를 돌리는 곳에 데이터를 저장한다.

아래서 쓰인 SQLite 관련 도큐멘테이션은 여기를 참고하자. 별도로 다루지는 않는다.

import sqlite3

# 일반적으로 connect는 host에 연결하는 행위이지만, SQLite은 곧바로 DB에 연결한다.
con = sqlite3.connect('ecommerce.db')

# 명령어를 실행해주기 위한 커서 선언
cur = con.cursor()

# event 테이블을 만들어준다. 일단 모두 text 타입으로 선언한다.
cur.execute("""
    CREATE TABLE events
        (event_time text, event_type text, product_id text, category_id text, category_code text, brand text, price text, user_id text, user_session text)
""")

import pandas as pd

# 아까 만든 데이터를 불러온다.
data = pd.read_csv('ecommerce_cosmetics_sampled_10000.csv')

lists = []
# 튜플 형태로 리스트에 넣어준다
for idx, row in data.iterrows():
    lists.append(tuple([*row.values]))

# executemany를 통해 한 번에 데이터를 넣는다
cur.executemany("""INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",lists)

# 트랜잭션을 수행하는 모든 함수는 반드시 커밋을 해줘야 실제 트랜잭션이 수행된다
con.commit()

# 연결을 닫는다
con.close()

이제 데이터가 생성되었는지 확인해 볼 차례다. 위에서 SQLite을 설치했다면, 터미널을 열고 해당 소스를 실행시킨 위치로 이동하자. 즉, ecommerce.db 파일이 생성된 위치로 가면 된다.

해당 위치에서 아래 명령어를 실행한다.

sqlite3 ecommerce.db #이제 sqlite3 DB에 접속했다

# sqlite에서는 시스템 명령어를 쓸 때 '.'을 앞에 붙인다
sqlite> .tables #전체 테이블 목록을 읽어온다
events #아까 생성한 events 테이블이 있어야 정상이다

sqlite> .headers on #출력 시 헤더를 보여주도록 설정을 바꾼다

sqlite> SELECT COUNT(*) FROM events; #테이블의 전체 행 개수를 불러온다
95806 #랜덤 샘플링이기 때문에 값은 약간씩 다르다

sqlite> SELECT * FROM events LIMIT 1;
# 결과가 출력된다

sqlite> .exit #혹은 .quit으로 쉘을 종료한다

데이터가 쌓인 것을 확인했다. 이번 글의 중점은 코호트 분석이므로 데이터를 탐색하고, 들여다보고, 빈 값을 채우는 등의 프로세스는 생략한다.

4) SQL로 데이터 기반 만들기

이제 실제 데이터를 만들어보자. 그 전에, 어떤 코호트 분석을 할 것인지 정하고 가자.

• 코호트의 유형은? 최초 상품 클릭(view) 기준 코호트
• 코호트의 크기는? 1주 단위
• 코호트의 측정 항목은? 유지율
• 코호트의 전체 기간은? 9주
• 코호트 집계 방식은? 일반적인 방식. 기준 롤링을 하지 않는다. 즉, 코호트 기준일에만 부합하면 집계한다.

데이터는 어떻게 만들까? 코호트 테이블을 행렬로 볼 때, 각 원소는 (기준 코호트, 기준 코호트로부터의 경과일)의 쌍이라고 할 수 있다.

즉, (코호트1, 코호트1 +1주), (코호트1, 코호트1 +2주)… 를 기준으로 그룹을 만든 것이 코호트 테이블이 된다.

con = sqlite3.connect('ecommerce.db')
cur = con.cursor()

result = con.execute("""
WITH 
base AS (
-- 문자열을 날짜로 바꿔주기 위한 용도
SELECT
    user_id,
    -- '주간'만 빼내는데, 연도가 바뀌면 계산이 틀어지기 때문에 현재 연도에서 가장 낮은 연도인 2019년을 뺀 만큼에 52를 곱한 값을 더해준다
    -- 즉, 2019년 마지막 주는 52가 되고, 2020년의 첫 주는 1 + (2020-2019)*52 = 53이 된다
    STRFTIME('%W', DATE(SUBSTR(event_time, 1, 10))) + (STRFTIME('%Y', DATE(SUBSTR(event_time, 1, 10))) - 2019) * 52 AS event_week
FROM events
WHERE event_type = 'view'
-- 9개의 주간으로 나누기 위해 기간을 제한해준다
AND STRFTIME('%W', DATE(SUBSTR(event_time, 1, 10))) + (STRFTIME('%Y', DATE(SUBSTR(event_time, 1, 10))) - 2019) * 52 <= 47
)
,first_view AS (
-- 우선 사용자별로 최초 유입 월을 찾는다. 이게 코호트가 된다.
SELECT
    user_id,
    MIN(event_week) AS cohort
FROM base
GROUP BY user_id
)
,joinned AS (
-- 기존 데이터에 위에서 찾은 코호트를 조인해준다. 그리고 기존 이벤트 월과 코호트 월의 차이를 빼준다
SELECT
    t1.user_id,
    t2.cohort,
    t1.event_week,
    t1.event_week - t2.cohort AS week_diff
FROM base t1
LEFT JOIN first_view t2
ON t1.user_id = t2.user_id
)

-- (기준 코호트, 기준 코호트로부터의 경과주) 쌍을 만들어 고유한 사용자 수를 센다
SELECT
    cohort,
    week_diff,
    COUNT(DISTINCT user_id)
FROM joinned
GROUP BY cohort, week_diff
ORDER BY cohort ASC, week_diff ASC
""").fetchall()

이제 데이터 추출이 끝났다. 판다스로 가져와서 행렬로 만들고, 추가적인 시각화까지 진행해보자.

5) 코호트 테이블 만들기

# 데이터프레임으로 만들고
# 컬럼의 이름을 바꿔주고
# 피벗 기능을 이용해 코호트 테이블 형태로 만들어준다
# 빈 값은 0으로 채운다
pivot_table = pd.DataFrame(result)\
    .rename(columns={0: 'cohort', 1: 'duration', 2: 'value'})\
    .pivot(index='cohort', columns='duration', values='value')\
    .fillna(0)

pivot_table

위 상태는 유지되는 고유 유저수를 보여주고 있기 때문에 비율로 바꾸어 주겠다. 또한 판다스에서 제공하는 background_gradient를 이용해 히트맵을 그려준다.

# 첫 번째 기간으로 나누어 비율로 만들어주고
# %가 나오도록 포맷팅을 해주고
# 색을 입혀준다

round(pivot_table.div(pivot_table[0], axis='index'), 2)\
    .style.format({k: '{:,.0%}'.format for k in pivot_table})\
    .background_gradient(cmap ='Blues', axis=None, vmax=0.2) 

6) 해석하기. 가져갈 수 있는 의미

이제 코호트 테이블을 읽어보자. 왼쪽의 39는 2019년의 39번째 주를 의미한다. 그리고 오른쪽의 duration에서 1~8은 코호트 주로부터 그만큼 경과한 주를 의미한다.

• 모든 코호트에서 1기간이 경과했을 때의 유지율은 평균 10%가 조금 넘는다.
• 코호트39에서 8기간이 경과했을 때의 유지율은 9%이다. 즉 39번째 주에 방문하여 상품을 본 고객 중, 8주 이후에도 똑같이 행동한 고객은 그 중 9%라는 의미이다. (단, 이 고객이 1~7주차 사이에도 그랬는지는 알 수 없다.)
• 3기간이 경과했을 때의 유지율은 4~8% 정도로 매우 낮은 편이다.

전반적으로, 유지율이 매우 낮은 것을 알 수 있다. 물론 상품을 본(view) 이벤트를 기준으로 하기 때문에 사이트 방문이라고는 볼 수 없지만, 일반적으로 이커머스에서 유지율(재방문율)은 낮아도 20% 정도인 업체가 많다. 이런 경우 대부분 유입 트래픽 광고를 엄청나게 뿌려대지만 당장 그 때만 고객이 유입하고 이후에 리타겟팅이 안 되는 경우가 많다.

또 어떤 데이터를 읽을 수 있을까? 테이블만 봐서는 사실 잘 와닿지 않는다. 우리는 이 분석을 통해 얻어가고자 하는 바를 미리 정의하지 않았기 때문이다. 미리 정의하지 않았기 때문에 코호트 분석을 통해 찾아내고자 하는 내용이 없었고, 그에 맞는 평가 지표도 선정되지 않았다.

7) 또다른 코호트 접근

이번에는 10월 한 달 동안 사용자가 본 브랜드별로 코호트를 만든다고 가정해보자. 마케터인 나는 브랜드별로 사용자의 성향이 다르고, 방문 성향도 다를 것이라고 생각한다. 따라서 브랜드에 따라 코호트를 나누었다.

result_2 = con.execute("""
WITH 
base AS (
-- 문자열을 날짜로 바꿔주기 위한 용도
SELECT
    user_id,
    STRFTIME('%W', DATE(SUBSTR(event_time, 1, 10))) + (STRFTIME('%Y', DATE(SUBSTR(event_time, 1, 10))) - 2019) * 52 AS event_week,
    event_type,
    brand
FROM events
-- 9개의 주간으로 나누기 위해 기간을 제한해준다
WHERE STRFTIME('%W', DATE(SUBSTR(event_time, 1, 10))) + (STRFTIME('%Y', DATE(SUBSTR(event_time, 1, 10))) - 2019) * 52 <= 47
AND brand IS NOT NULL
AND event_type = 'view'
AND DATE(SUBSTR(event_time, 1, 10)) >= '2019-10-01'
AND DATE(SUBSTR(event_time, 1, 10)) <= '2019-10-31'
)
,first_view AS (
SELECT
    user_id,
    brand AS cohort,
    MIN(event_week) AS cohort_time
FROM base
GROUP BY user_id, brand
)
,joinned AS (
SELECT
    t1.user_id,
    t2.cohort,
    t1.event_week,
    t1.event_week - t2.cohort_time AS week_diff
FROM base t1
LEFT JOIN first_view t2
ON t1.user_id = t2.user_id
AND t1.brand = t2.cohort
)

SELECT
    cohort,
    week_diff,
    COUNT(DISTINCT user_id)
FROM joinned
GROUP BY cohort, week_diff
ORDER BY cohort ASC, week_diff ASC
""").fetchall()

# 데이터프레임으로 만들고
# 컬럼의 이름을 바꿔주고
# 피벗 기능을 이용해 코호트 테이블 형태로 만들어준다
# 빈 값은 0으로 채운다
pivot_table_2 = pd.DataFrame(result_2)\
    .rename(columns={0: 'cohort', 1: 'duration', 2: 'value'})\
    .pivot(index='cohort', columns='duration', values='value')\
    .fillna(0)\
    .sort_values(by=[0], ascending=False)\
    .iloc[:10, :]

# 상위 10개만 잘랐다
pivot_table_2

# 첫 번째 기간으로 나누어 비율로 만들어주고
# %가 나오도록 포맷팅을 해주고
# 색을 입혀준다

round(pivot_table_2.div(pivot_table_2[0], axis='index'), 2)\
    .style.format({k: '{:,.0%}'.format for k in pivot_table_2})\
    .background_gradient(cmap ='Blues', axis=None, vmax=0.2) 

이번에는 어떨까? 브랜드 코호트마다 확실히 유지율이 다른 것을 알 수 있다. 특정 브랜드를 찾았던 고객은 계속해서 방문을 하고, 다른 브랜드들은 그렇지 않다. 여기에는 브랜드의 속성이 주는 재구매주기라든지 여러 요소들이 작용할 것이다. 브랜드를 떠나 화장품의 카테고리별로 보는 것도 좋은 분석이 될 수 있다.

4. 마치며

코호트 분석에 대해 정리해두었더 내용들을 글로 옮길 수 있어 마음이 편안하다. 😃 아무쪼록 누구에게든 도움이 되었으면 좋겠고, 앞으로도 데이터 분석과 관련한 글들을 계속해서 작성할 예정이다!

생각중인 주제로는 퍼널 분석, RFM 세그멘테이션, LTV, 전환기여(FirstTouch, LastTouch) 등이 있는데, 다양하게 다루어보도록 하겠다 :)

1. 아파치 에어플로우란?

아파치 에어플로우는 Python 기반으로 태스크를 등록하고, 워크플로우를 관리하고, 모니터링하는 도구이다. 간단하게 기능을 요약하자면 다음과 같다.

• 특정 워크플로우(DAG이라고 한다)를 간편한 파이썬 코드를 통해 등록할 수 있다.
• DAG 사이의 순서나, 조건부 실행 등 복잡한 플로우를 설계할 수 있다.
• 또한 Operator를 통해, 다른 프레임워크들과의 통합을 간단하게 해두었다.
• 등록한 DAG의 스케줄링이 세부적으로 가능하다.
• 스케일링이 가능하다.
• 간편한 WebUI를 통해 모니터링과 DAG 관리가 가능하다.

기타 태스크 관리 도구나 단순한 Crontab을 통해 태스크를 관리했던 사람이라면 에어플로우가 가져다주는 이점이 매우 크다는 것을 알 것이다. 일단 최초 세팅이 끝나고 나면, DAG만 기계적으로 추가해주면 되기 때문이다.

하지만, 이 편리함과 더불어 에어플로우를 경험하며 겪게 되는 불편함들도 있는데, 이러한 부분들은 아래에서 다루도록 하겠다.

2. 개념 및 동작 원리 소개

에어플로우에서 주로 쓰이는 개념은 다음과 같다.

DAG

Directed Cyclic Graph의 약자로, 순환하지 않고 방향이 있는 그래프라고 생각하면 된다. 아래 그림은 여기서 따왔다.

각 DAG은 에어플로우의 핵심 개념으로써 관련 태스크들을 선언하고, 묶고, 그들간의 흐름을 정리하고, 동작에 관련된 메타 정보를 선언해주는 공간이다.

단일 파이썬 파일 하나가 DAG에 대응되는 개념이며, 해당 파일 내에서 선언해주는 방식으로 진행된다.

자세한 DAG의 선언에 관한 내용은 여기를 읽어보자. 단, 왼쪽 메뉴의 버전을 꼭 확인하자.

Task & Operator

태스크는 DAG 내부에서 실행되는 가장 기본적인 단위이다. 단일 태스크, 또는 태스크들이 모여 DAG을 구성하고, 그들 사이에 의존성이나 실행 규칙 같은 것들을 선언할 수 있다.

태스크의 종류는 크게 세 가지가 있다.

• Operator: 에어플로우 자체적으로, 또는 프로바이더가 제공해주는 템플릿화된 태스크를 의미한다. 예를 들어 EmailOperator를 사용하면 이메일에 관한 기능을 직접 구현할 필요 없이 쉽게 가져다 쓸 수 있고, MySqlOperator를 사용하면 MySQL에 손쉽게 연결하여 필요한 기능을 쓸 수 있다.
• Sensor: 오퍼레이터와 비슷하지만, 외부의 이벤트가 발생할 때까지 기다리는 태스크를 의미한다.
• TaskFlow: 데코레이터 @task 형태로 쓰이며, 사용자 지정 파이썬 함수를 의미한다.

여기서 에어플로우의 편리함이 보이는데, 정말 무수한 종류의 Operator를 통해 편하게 기능 구현을 할 수 있다는 점이다. 해당 목록은 여기서 살펴보자. 카산드라, 드루이드, 하이브, AWS, Azure, GCP 등 수많은 플랫폼에 연결이 가능하다.

동작 원리

hello_world라는 DAG이 있다고 가정하자. DAG은 아래처럼 3개의 태스크로 구성되어 있다. 문법은 고려하지 않고, 그냥 어떤 형태로 되어있는지만 참고하자.

hi = make_hi()
ready_to_say_hi = ready_hi(hi)
say_hi(ready_to_say_hi) 

이때, 에어플로우는 대략적으로 아래와 같은 과정을 거친다.

1. 스케줄러가 DAG의 선언을 감지하고, 메타DB에 반영한다.
2. 스케줄러가 해당 시간에 동작해야 하는 hello_world DAG을 발견하고, executor에 전달한다.
3. executor가 단일 executor로 동작하거나, 분산 처리가 가능할 경우 worker들에게 배분한다.
4. 최초로 make_hi 태스크를 실행한다. 이때 결과값을 hi라는 변수에 저장하는데, 이를 로컬 머신이 아닌 메타DB에 저장한다. (XCom)
5. ready_hi가 메타DB에 있는 hi를 다시 가져와서 다음 태스크를 실행한다. 마찬가지로 결과를 또 저장한다.
6. 똑같은 원리로 say_hi를 실행한다.
7. 실행 결과를 메타DB에 저장하고, DagRUN을 업데이트한다.

동작에 관한 내용은 이 정도로 하고, 사용하며 겪은 단점과 주의점을 살펴보도록 하겠다. (태스크 관리에 대해 에어플로우만큼의 세분화된 기능을 제공해주는 툴은 사실상 없다고 보기 때문에 장점은 따로 다루지 않겠다. 그냥 전반적인 기능이 주는 편리함!)

3. What airflow IS NOT! 에어플로우에 대한 오해

일단 내가 에어플로우에 가지고 있던 잘못된 오해들을 먼저 다루겠다.

1. 에어플로우는 워크 플로우를 분배해주지, 비동기 분산처리를 해주는 시스템이 아니다.

에어플로우를 도입하기 전에는 Celery를 기반으로 자체적인 태스크 시스템을 구축하여 사용하고 있었다. 이때는 자연스럽게 비동기 호출을 통해 태스크를 비동기로 무수히 많이 찍어냈고, 그런 동작을 사용함에 있어 어려움이 없었다.

그런데 에어플로우로 마이그레이션을 하고 나서, 도통 구현하기 어려운 부분들이 있었다. 예를 들어, 특정 스트림 파이프에서 데이터를 퍼내는 작업을 진행할 때 나는 이 파이프를 빠르게 비워내기 위해 비동기로 여러 워커를 돌리고 싶었다. 예를 들어 데이터가 100덩어리가 있으면 1덩어리씩 넘어가면서 비동기 태스크를 100번 호출하는 식으로 구현을 하려 했는데, 이러한 동작이 불가능했다.

그 이유는 에어플로우의 태스크는 미리 정의된 방식대로 동작하기 때문이다. 다이나믹하게 태스크를 선언하는 것은 기술적으로 가능은 하지만, 개별 태스크 관리가 어려워진다. 동적으로 선언을 했을 때, 실패 시 동작은 어떻게 할 것이며 나중에 WebUI에서 로그를 살필 때도 태스크가 사라질 수 있기 때문에 문제가 된다.

추가적으로, 에어플로우는 자체적으로 비동기 실행을 지원하지 않는다. 다른 파이썬 라이브러리를 통해서는 직접 구현해야만 한다. 비동기를 지원하지 않는 이유는 비동기 태스크들이 허용된다면 정기적으로 돌아야 하는 스케줄된 태스크들의 실행에 영향을 줄 수 있기 때문일 것이다. (동작 slot을 차지하여 제 시간에 실행이 안 되는 등..)

2. 에어플로우는 이벤트 기반 동작을 하지 않는다.

에어플로우는 이벤트를 기반으로 동작하지 않는다. 즉, 서버의 역할처럼 대기를 하고 있다가 특정 요청이 들어왔을 때 태스크를 수행하는 용도가 아니다. (어찌어찌 그렇게 구현할 수는 있겠지만)

에어플로우는 철저히 스케줄링을 위한 시스템이다.

3. 에어플로우는 데이터를 주고받기 적합한 시스템이 아니다.

위에서 설명한 XCom 시스템을 통해 에어플로우는 데이터를 주고받는다. 알아두어야 할 점은, 파이썬에서 하듯이 태스크의 결과값을 변수에 저장하는 행위 자체가 메타DB에 접근하는 일이라는 사실이다.

즉, A 태스크에서 B 태스크로 변수를 넘기면, 이 정보는 로컬 머신에서 전달되는 것이 아닌, 메타DB를 거쳐서 나온다. 따라서 변수 자체가 이러한 트랜잭션을 늘리는 행위이고, 더 나아가 큰 데이터를 주고받는 순간 에어플로우 자체에서 통신 제한에 걸리거나 그 외 성능 저하 이슈를 겪을 수 있다.

이는 애초에 워커가 분산되어있는 시스템을 가정하고 있기 때문에 데이터의 중재소로써 메타DB를 사용하기 때문이다.

4. 에어플로우의 단점

이번에는 에어플로우의 단점을 살펴보자.

1. 잔버그가 많다. 불친절하다.

에어플로우 사이트에 들어가면 상당히 문서화가 잘 되어 있음을 알 수 있다. 하지만 동시에, 에어플로우를 깊게 파다보면 문서화가 드문드문 되어있고 버전에 잘 맞지 않는 부분들이 있다는 것도 알 수 있다.

더 나아가, 에어플로우를 사용하다보면 알 수 없는 DAGRUN 실패나 메타DB 관련 버그가 종종 일어난다. 특히 메타DB 관련 버그는 고치기도 어렵기 때문에 초기화가 가능한 상태면 그냥 초기화하곤 했다.

물론 에어플로우에 대한 이해가 부족하여 버그를 더 많이 겪은 걸 수 있지만, 에어플로우 깃헙을 들어가보면 2.x 이상의 버전에서 에러 케이스가 많고, 현재 알려진 버그도 상당히 많다.

2. 워커간에 코드 동기화가 번거롭다.

에어플로우 자체적으로 워커 A와 B 사이의 코드를 동기화해주는 기능이 없다. 따라서 깃이나 기타 배포 서비스를 통해 두 워커의 DAG 코드를 일치시켜주어야 한다.

3. (놀랍게도) 로그 관리를 직접 해주어야 한다.

에어플로우는 크게 두 종류의 로그를 만들어낸다.

1. 메타DB에 쌓이는 로그
2. 로컬 머신에 쌓이는 로그

정말 놀랍게도, 이 두 로그는 자동으로 로테이트되지 않는다. 기간이 지나면 자동으로 삭제되지 않고, 사용자가 직접 제거해주어야 한다.

1번의 메타DB는 airflow-maintenance-dags라는 커스텀 DAG을 사용하면 되고, 2번은 logrotate 와 같은 툴을 사용하면 된다.

로그를 제거하지 않으면 용량이 터지는 경우가 무조건 발생하기 때문에 꼭 사전에 세팅해주고, 잘 제거가 되는지 확인해주자.

4. 태스크가 많아지면 메타DB와 WebUI가 버벅거린다.

예전에 고객사별로 태스크를 관리할 필요가 있었는데, 그렇게되면 개별 DagRUN에 200~300개의 태스크가 생기곤 했었다. 이 경우, 돌고있는 수만큼 태스크가 메타DB에 연결하기 때문에 메타DB의 부하가 커지고, 이를 UI로 확인하려고 해도 WebUI도 버벅거리게 된다.

이는 단점이라기보다도 에어플로우의 사용에 적합하지 않은 경우가 아니었나 생각하지만, 그래도 생각난 김에 적어보았다.

5. 마무리하며

에어플로우는 분명히 좋은 툴이다. 간단한 코드 몇 줄로 태스크 사이의 의존성을 나타낼 수 있고, 스케줄링, 로그까지 남겨주기 때문에 간편하다. 다만 사용함에 있어 꼭 알아야 할 부분들이 잘 전달되어 있지 않은 것도 사실이다.

내가 에어플로우를 처음 접할 때 너무나도 궁금했던 내용들이었는데, 쉽게 정보를 구할 수 없었다. 그동안의 의미있는 삽질을 통해, 구글링을 통해, 에어플로우 슬랙을 통해 알게된 내용들을 이렇게 정리하여 공유할 수 있어 기쁘다. 😙

1. 들어가기에 앞서

일단 지난 시간에 아랑고DB에 넣어둔 데이터를 확인해보자. 보다 복잡한 분석을 위해서는 파이썬을 통해 아랑고DB에 AQL을 날려서 데이터를 주고받고 해야하지만 여기서는 최대한 간단한 형태의 추천 시스템을 실습할 것이기 때문에 오로지 AQL을 통해서만 결과를 낼 것이다.

아랑고DB 웹 UI로 들어가서 movie_ratings 데이터베이스를 선택한 후, 좌측의 QUERIES 메뉴를 클릭하자.

웹UI 관련 설정이 기억나지 않거나 이슈가 있으면 이 글에서 Web UI 세팅 부분을 다시 살펴보자.

2. 데이터 살펴보기

이제 실제로 데이터가 잘 들어갔는지 확인해보자.

우선 각 컬렉션의 데이터 개수를 확인한다. COLLECTIONS에서 각 컬렉션을 들어가서 좌측 하단의 데이터 개수를 확인해도 되는데, 여기서는 AQL을 통해 확인해보겠다.

// AQL은 하나의 리턴만 하기 때문에 아래를 각각 실행시켜줘야 함에 주의
RETURN length(rated)  // 84942

RETURN length(User) // 1000

RETURN length(Movie) // 17763

유저는 천 명으로 제한했었기 때문에 딱 1000개의 레코드가 있고, 영화는 17763개, 그리고 평가는 84942개가 있다.

이번에는 유저별, 영화별, 년도별 평가의 개수를 확인해보자.

// AQL에서 통계값을 뽑는 관련 함수는 없는 것 같다. 각각을 개별 계산해준다. 
LET rates = (
FOR e IN rated
    COLLECT
        user = Document(e._from).user
    AGGREGATE
        cnt = COUNT(1)
    RETURN {
        user,
        cnt
    }
)

RETURN {
    min: min(rates[*].cnt),
    max: max(rates[*].cnt),
    avg: avg(rates[*].cnt)
}

위 쿼리는 일단 유저별로 평가의 개수를 모아준 뒤에(rates), 이 배열에서 cnt의 최소, 최대, 평균값만을 다시 계산한 결과이다.

유저들은 최소 20개에서 최대 2500개(!)의 평가를 내렸고, 평균적으로 84개의 평가를 내렸다. 아마 애초에 Netflix Prize에 사용되었던 데이터이기 때문에 양질의 평가를 한 사람들만을 추려놓은 것이라 생각된다.

// 영화별 rating 개수
FOR e IN rated
    COLLECT
        movie = Document(e._to).name
    AGGREGATE
        cnt = COUNT(1)
    
    SORT cnt DESC
    
    RETURN {
        movie,
        cnt
    }

쿼리는 아까와 유사하다. 그룹을 해주는 기준만 영화 타이틀로 바뀌었을 뿐이다. 같은 통계를 내보면, 최소 1개의 평가가 있는 영화부터, 최대 640개의 평가가 있는 영화까지 다양하다. 평균적으로 22개의 평가가 있다.

오해하면 안 될 것이, 전체 Netflix의 평가가 아닌, 내가 임의로 선별한 아주 작은 데이터가 그렇다는 것이다.

// 이번에는 년도별로
FOR e IN rated
    COLLECT
        date = SUBSTRING(e.date, 0, 4)
    AGGREGATE
        cnt = COUNT(1)
    
    SORT cnt DESC
    
    RETURN {
        date,
        cnt
    }

얘는 따로 통계를 볼 필요가 없다. 위 쿼리만 실행해도 2000년~2005년 사이 년도별 개수가 나오기 때문이다. 확실히 오래된 데이터라 2005년 데이터가 가장 많다.

3. 협업 필터링(Collaborative Filtering, CF)

이번 실습에서는 추천 시스템의 한 가지 방법인 협업 필터링을 사용한다. (간단히 CF라고 줄여서 부르겠다)

CF는 가장 널리 사용되는 기법 중 하나이고, 이름 그대로 주변 사람들의 정보를 이용하여 (협업, collaborate) 예측을 만들어내는 (filtering) 기법이다.

위키피디아의 예시를 가져와보면, CF의 기본 아이디어는 아래와 같다.

• 유저 A와 유저 B가 1번 주제에 대해 의견이 같다고 가정하자.
• 그럼 A는 임의의 유저보다, 또다른 2,3,4.. 주제에 대해서도 B와 의견이 같을 확률이 높을 것이다.

즉, 비슷한(similar) 유저일수록 취향이 비슷할 것이고, 그 점을 기반으로 추천 값을 만들어내는 것이다.

CF에도 여러가지 종류가 있는데, 크게는 메모리 기반(Memory-based or Neighborhood-based)과 모델 기반(Model-based)로 분류된다.

• Memory based CF : ‘메모리’ 기반에서는, 메모리에 주어진 ‘데이터’ 내에서 여러 계산을 거쳐 추천 값을 생성한다.
• Model based CF : ‘모델’ 기반에서는, 주어진 ‘데이터’를 통해 모델을 학습시켜서 이 모델이 추천 값을 생성하도록 한다.

추천 시스템에 관해 공부하고, 설명할 내용은 정말 많이 있다. CF도 자세하게 들어갈 수 있고, 이 외의 추천 방법론, 그리고 평가가 implicit 할 때의 방법 등.. 다룰 내용이 많다. 이건 기회가 되면 따로 시리즈로 작성해보도록 해야겠다. 😃

4. 메모리 기반 CF

여기서는 메모리 기반 CF를 쓸 건데, 메모리 기반도 크게 두 가지 종류로 나뉜다.

A 유저의 아이템 B에 대한 평가를 예측한다고 할 때, 각각은 아래처럼 해석할 수 있다.

• User-based : A 유저와 비슷한 성향을 가진 다른 유저들이 B에 대해 어떤 평가를 내렸는지 종합해본다.
• Item-based : B 아이템과 비슷한 성향을 가진 다른 아이템들에 대해 A가 어떤 평가를 내렸는지 종합해본다.

여기서는 user-based CF를 사용할 예정이다. 다양한 가정에 따라 다양한 수식을 적용할 수 있으니 함께 찬찬히 살펴보자.

1) CF with equal similarity

가장 일반적인 형태는 아래와 같다.

[r_{u,i} = \frac{1}{N}\sum_{u \prime \in U}r_{u \prime, I}]

위 수식의 좌변은 유저 $u$가 아이템 $i$에 대해 내릴 평가를 의미한다. 이는 예측값이다.

우변은 타겟 유저와 유사한 성향을 가진 집합 $U$에 속한 $u\prime$ 들이 상품 $i$에 대해 내린 평가를 다 더한 뒤, 집합의 크기인 N으로 나누어주고 있다. 즉, 타겟 유저의 아이템 $i$에 대한 평가를 예측하기 위해, 유사한 상위 N명의 $i$에 대한 평가를 평균낸 것이다. ($i$에 대해 평가한 사람들에 한해서 걸러내기 때문에 각 아이템마다 이 그룹은 달라진다)

2) CF with different similarity

여기서 의문을 가질 수 있는 부분이 있다. 유사한 상위 N명에 대해 그들의 점수를 평균내는 것은 이해하겠는데, 이들 중 누군가는 정말 비슷한 취향을 가졌을 것이고, 누군가는 그다지 취향이 맞지 않을 수도 있다.

따라서, 유사 그룹의 점수를 평균 내는 것이 아닌, 유사도에 따라 가중 평균을 해줄 수 있다.

[r_{u,i} = k \sum_{u \prime \in U}sim(u, u\prime)r_{u\prime, i}]

위 식에서 실제로 k값은 $ 1/ \sum_{u\prime \in U}|sim(u, u\prime)| $에 해당한다. 즉 가중 평균이기 때문에 가중치들의 합으로 나눠주는 것이다.

3) CF with different similarity, rating adjusted

더 나아가, 평가 점수 자체에 대한 의문을 제기할 수 있다. A 유저는 점수를 후하게 주는 사람이고, B는 그렇지 않은 사람일 때, A의 4점과 B의 2점은 비슷한 평가 수준일 수 있다. 그럼, 개인이 점수 주는 범위를 계산해서, 잘 스케일링 할 수 있지 않을까?

[r_{u,i} = \bar r_u + k \sum_{u \prime \in U}sim(u, u\prime)(r_{u\prime, i} - \bar r_{u\prime})]

2번 식과 달라진 부분은, 유사 사용자의 평가를 그대로 쓰는 대신, 평균 중심화(mean-centered) 된 값을 가중평균하고, 거기에 타겟 사용자의 평균 점수를 다시 더해준다는 점이다.

즉, 사람마다 평가 기준이 다르기 때문에 이를 나름대로의 표준화 과정을 거친 상태에서 가중 평균을 해주고, 거기에 다시 타겟 사용자의 평균값을 더해준다는 의미이다.

4) CF in graph

CF는 구현 방식이 정말 다양하기 때문에 그래프DB의 특성을 살려서 구현을 할 수도 있다. 예를 들어, ‘유사도’라는 개념을 ‘A 사용자 노드에서 k 스텝을 랜덤 워크 했을 때, 많이 겹치는 정도’라고 정의할 수도 있다. 즉, A 사용자에서 출발해서 임의의 그래프 횡단을 했을 때, 특정 사용자에게 많이 도달할수록 유사한 타겟이라는 의미가 된다.

Disclaimer

다만, 여기서는 1번 수식을 채택한다. 모두가 동일한 가중치를 가지고 있다고 가정하고, 가장 간단한 형태의 계산만을 할 것이다. 아랑고DB에 중점을 둔 글이기 때문에 이렇게 문법을 쓸 수 있다는 것에 집중하고, 추천 시스템의 정확한 프로세스와 검증 과정은 추천 시스템 관련 글에서 다루도록 하겠다!

5. 그래프를 통한 유사도 계산

이제 실전으로 넘어가보자!

유사도를 계산하는 방법은 피어슨 상관계수를 쓰거나, 코사인 유사도를 쓰는 방법 등이 있다. 여기서는 더 구현이 편한 코사인 유사도를 사용한다.

[cos(\vec x, \vec y) = \frac{\sum_{i\in I_{xy}}r_{x,i}r_{y,i}}{\sqrt {\sum_{i \in I_x}r_{x, i}^2} \sqrt {\sum_{i \in I_y}r_{y, i}^2}}]

위 수식들을 그래프DB로 되살려보자. 여기서는 전체 데이터셋에 대한 추천값을 계산하지 않고, 임의의 타겟 사용자 1명에 대해, Top-K 영화를 추천해 볼 것이다. 유사 그룹은 30명 사이즈라고 가정하겠다.

LET target = Document('User/1158991')

LET sim = (
	//1
    LET movies_target = (
    FOR movie, e, p IN OUTBOUND target rated
        RETURN e
    )
	
	//2
    LET movies_target_distinct = FLATTEN(UNIQUE(
        RETURN movies_target[*]._to
    ))

	//3
    LET movies_target_ratings_square = (
        RETURN SUM(movies_target[* RETURN POW(CURRENT.rating, 2)])
    )	
	
	//4
    LET users_with_same_movie = (
    FOR movie IN OUTBOUND target rated
        FOR other_user IN INBOUND movie rated
            FILTER target != other_user
            RETURN DISTINCT other_user
    )
	
	//5
    FOR other_user IN users_with_same_movie
        LET movies_other = (
        FOR movie, e, p IN OUTBOUND other_user rated
            RETURN e
        )
        LET movies_other_distinct = FLATTEN(UNIQUE(
            RETURN movies_other[*]._to
        ))
        LET movies_other_ratings_square = (
            RETURN SUM(movies_other[* RETURN POW(CURRENT.rating, 2)])
        )
        LET dot_product = SUM(
        FOR movie IN INTERSECTION(movies_target_distinct, movies_other_distinct)
            LET score_target = movies_target[* FILTER CURRENT._to == movie].rating
            LET score_other = movies_other[* FILTER CURRENT._to == movie].rating
            RETURN score_target * score_other
        )
        
        LET cosine_sim = dot_product / (SQRT(movies_target_ratings_square) * SQRT(movies_other_ratings_square))
        
        SORT cosine_sim DESC
        
        RETURN {
            other_user,
            cosine_sim
        }
)

위 쿼리는 일단 타겟 사용자(target)을 임의로 지정해준다. 그리고 이 타겟과 나머지 사용자에 대한 유사도를 계산해주고 있다.

• 1: 타겟 사용자가 평가한 영화와 그 엣지를 모은다
• 2: 타겟 사용자가 평가한 고유 영화를 모은다
• 3: 타겟 사용자가 평가한 점수의 제곱의 합에 루트를 씌운 값. 즉, 코사인 유사도에서 $ \lVert \vec x\rVert $ 에 해당하는 값
• 4: 2번에서 모은 영화를 함께 본 사용자의 목록. 어차피 함께 본 영화가 없다면 유사도가 없기 때문에 이들은 쿼리의 효율성을 위해 신경쓰지 않는다
• 5: 4번에서 대상 유사 그룹에 대해, 그들의 $ \lVert \vec y \rVert$ 값과, 타겟과 그들의 스칼라 곱($ \vec x \cdot \vec y $)을 구한다

그리고 이어서 다음 쿼리가 진행된다 (글에서는 분리했지만, 위 쿼리에 바로 이어서 실행된다)

FOR movie IN Movie
	// 1
    LET inflow = FIRST(
    FOR v, e, p IN INBOUND movie rated
        COLLECT WITH COUNT INTO length
        RETURN length
    )
    
    FILTER inflow >= 30
    	
	// 2
    LET candidates = (
    FOR user IN INBOUND movie rated
        FILTER user != target
        RETURN DISTINCT user
    )

	// 3    
    LET predict_score = AVG(
    FOR doc IN sim
        FILTER doc.other_user IN candidates
        LIMIT 30
        FOR v, e, p IN OUTBOUND doc.other_user rated
            FILTER v == movie
            RETURN e.rating
    )
    
    SORT predict_score DESC
    
    RETURN {
        'name': movie.name, 
        predict_score
    }

이제 타겟 사용자의 전체 영화에 대한 예측 평점을 구할 차례이다.

• 1: 전체 영화를 순회하되, 유의미한 선별을 위해 평가가 30개 이상인 영화만 남긴다 (평가가 없는 cold start 문제는 다른 방법으로 해소해 줘야 한다! 여기서는 무시함)
• 2: 해당 영화를 본 사람들의 목록을 만든다
• 3: 처음에 구한 (타겟, 다른 유저) 유사도를 순회하며, 함께 이 영화를 본(2번 목록에 있는) 사람들의 점수를 평균낸다. 유사 그룹은 30명으로 제한한다.

이렇게 결과값을 내면 아래 그림처럼 타겟이 좋아할만한 영화의 점수와 순위를 매길 수 있다!

이런 식으로 전체 유저에 대해 적용하면, 가장 단순한 형태의 CF를 구현해 볼 수 있다. 단, 그래프DB를 통해 더 깊이 들어갈 예정이라면 아래 6번은 꼭 읽어보자!

6. 그래프DB를 CF에 적용할 때의 어려움, 주의점

메모리 기반 CF에서 계산의 핵심은 유사도를 구하는 작업이다. 사용자와 사용자, 혹은 아이템과 아이템 사이의 유사도를 구해야하는데, 이는 $O(m^2)$의 복잡도를 갖는다. 따라서 대상의 수가 늘어날수록 급격하게 계산의 비용이 높아진다.

사실 그래도, 작은 규모의 데이터에서는 파이썬의 행렬 곱을 통해 유사도 행렬을 쉽게 구할 수가 있는데, 그래프DB에서는 그렇지가 않은 것 같다. 그래서 위 AQL도 반복적인 계산이 많고, 비효율적으로 구성된 것 같다. (분명 코드 개선의 여지는 있다! 근데 여기선 그게 중점이 아니어서 일단 넘어감 😂)

결론적으로, 일반적인 메모리 기반 CF를 구현하는 게 목적이라면, 그냥 파이썬을 사용하기를 추천드린다. 그래프DB를 활용하려면 유사도와 같이 계산 집약적인 작업은 파이썬으로 넘기고, 탐색 관련 작업을 그래프로 구현하는 것이다. 그렇게 하려면 그래프 초기 설계부터 잡고 들어가는 게 맞다. 즉, 사용자 A와 B 사이에 유사도 엣지를 추가하는 등의 방법을 쓰는 것!

7. 아랑고 시리즈를 마치며

이 글을 포함하여 무려 11편의 아랑고DB 시리즈가 끝이 났다. 아랑고DB를 접한 뒤, 꼭 그래프DB에 대한 가이드를 만들고 싶었기에 꾸준히 작성해왔는데, 돌아보니 시작하길 잘했다는 생각이 든다 :)

Neo4j가 주류인 상황에서 아랑고DB를 검색하는 한국 사용자가 얼마나 될까 싶지만, 그래프DB에 대한 간단한 개념을 익히기에는 아랑고DB만한 게 없다고 생각한다. 누군가에게는 가뭄의 단비처럼… 도움이 되길 바란다.

아랑고 시리즈를 읽어주셔서 감사하다는 말씀을 전하며, 저는 또 다른 글로 찾아 뵙도록 하겠습니다! 감사합니다.

1. 아랑고DB란? 왜 쓰는가?
2. 아랑고DB 세팅하기 on Ubuntu
3. 아랑고DB 쉘로 붙어서 명령어 체험해보기, 실체 파악해보기
4. AQL(Arango Query Lang) 배워보기 1
5. AQL(Arango Query Lang) 배워보기 2 - RETURN / UPDATE
6. AQL(Arango Query Lang) 배워보기 3 - REPLACE / UPSERT / REMOVE
7. 그래프 개념잡기
8. 그래프 횡단하기 Graph Traversal
9. 데이터 모으기 COLLECT / AGGREGATE / MIN_BY, MAX_BY
10. 프로젝트. 그래프를 통한 영화 추천시스템 만들어보기 1
11. (지금 보고있는 글) 프로젝트. 그래프를 통한 영화 추천시스템 만들어보기 2 (최종편)

1. Kaggle이란

Kaggle 캐글은 데이터를 통해 현실의 문제를 해결하기 위한 플랫폼이다.

데이터셋을 제공하고, 상금을 걸어 세계 각지에 있는 소중규모 팀들이 챌린지를 하는데,** 원하는 결과값에 가장 가까운 팀이 상금을 차지하는 방식이다.

이러한 챌린지뿐만 아니라 수많은 데이터셋을 제공하고, 이러한 데이터셋을 다른 사람들은 어떻게 접근했고 분석했는지도 살펴볼 수 있다.

나는 그 중에서 Netflix Prize Data라는 데이터셋을 가져와 변형하였다.

내 글을 참고하여 자신만의 데이터셋을 찾아 맵핑해도 좋으나, 이러한 과정이 처음이라면 내 글을 그대로 따라하는 것을 추천한다.

나는 1 ~ 4번의 전체 데이터 파일에서, 1번 파일만 사용하였고, 그 중에서도 1000명의 유저에 대한 데이터만 샘플링하였다.

최종적으로는 아래 두 개의 파일로 정리했으니, 받아서 사용하면 된다.

• 영화에 관한 메타 데이터 파일 : 여기
• 사용자 평가에 관한 데이터 파일 : 여기

혹시 다운로드가 안 된다면 여기에 들어가서 직접 다운로드 해주자.

두 파일을 로컬에 저장해도 좋고, 그냥 url을 코드에 넣어서 그때그때 다운로드해도 무방하다.

2. 데이터 살펴보고 변형하기

일단 데이터를 불러와서, 최종 형태만 살펴보자.

데이터는 csv 형태이므로 pandas 패키지의 read_csv를 사용한다.

import pandas as pd

movies = pd.read_csv('https://raw.githubusercontent.com/ud803/ud803.github.io/main/public/movie_titles.csv', encoding='ISO-8859-1', header=None, names=['id', 'year', 'name'])
ratings = pd.read_csv('https://raw.githubusercontent.com/ud803/ud803.github.io/main/public/netflix.csv', encoding='ISO-8859-1', header=None, names=['user', 'movie', 'rating', 'date'])

print(movies.shape, ratings.shape)
#(17770, 3), (84942, 4)

shape 함수를 통해 데이터프레임의 사이즈를 보면, 각각 17770, 84942개의 행을 가진 것을 알 수 있다.

데이터를 더 자세히 살펴보자

# 타입 확인하기
movies.dtypes
ratings.dtypes

# NA값 확인하기
movies[movies.isna().any(axis=1)]
ratings[ratings.isna().any(axis=1)]

확인해보니 movies의 출시 연도가 float 형태로 되어있다.

정수형으로 바꾸고 NA 값도 날려주자.

# dropNA
movies = movies.dropna()

# 제작 연도를 정수형으로 바꿔준다
movies = movies.astype({'year' : 'int32'})

3. 아랑고에 넣을 데이터 설계하기

이제 가장 재미있는 일만 남았다. 이 데이터들을 어떻게 아랑고DB에 맵핑할 것인가?

모든 데이터베이스가 마찬가지이지만, 일단 내가 데이터를 어떤 형태로 저장할 것이고, 어떻게 참조(쿼리)할 것인가를 구상해야 한다.

내 계획은 이렇다. 영화 추천 시스템이니까 당연히 영화(M)라는 버텍스가 있어야 하고, 사람(U)을 나타내는 버텍스가 또 있어야 한다.

그리고 사람(U)이 영화(M)를 평가(R)하는 엣지가 중간에 존재해야 한다. TMI이지만 Neo4j Cypher에서는 이를 (User) -[rated]-> (Movie) 와 같은 직관적인 방식으로 표현한다.

거기에 더해서, 나는 영화의 출시 연도도 추천 시스템에 넣고 싶다. 결과적으로, 아래의 같은 5개의 컬렉션들이 생성되어야 한다.

• 버텍스 컬렉션 : 영화(Movie), 사람(User), 출시연도(Year)
• 엣지 컬렉션 : (사람)-[rated]->(영화), (영화)-[created_at]->(출시연도)

4. 아랑고에 데이터 맵핑하기

자 이제 데이터를 어떻게 넣으면 될까? 이제껏 배운 방식을 총동원하여 각각에 맞는 데이터를 넣어줘도 되지만, 우리는 이미 만들어진 패키지를 가져다 쓰면 된다.

바로 python-arango-mapper라는 패키지이며, 내가 만들었다.. 😊

맵핑할 데이터에 대한 스키마를 지정해주면, 자동으로 데이터를 넣어준다.

우선 pip을 통해 패키지를 설치해주자.

# Python 3.7 이상에서 동작한다!!

pip install python-arango-mapper

schemas = {
# Type 1
    'Movie': {
        'type': ('vertex', 'unique_vertex'),
        'collection': 'Movie',
        'unique_key': ('id',),
        'fields': {
            'name': 'name',
            'year': 'year'
        },
        'index': [
            {'field' : ('name',), 'unique' : False, 'ttl' : False}
        ]
    },
    # Type 1
    'Year': {
        'type': ('vertex', 'unique_vertex'),
        'collection': 'Year',
        'unique_key': ('year',),
        'fields': {
            'year': 'year'
        },
        'index': [
            {'field' : ('year',), 'unique' : True, 'ttl' : False}
        ]
    },
    # Type 1
    'User': {
        'type': ('vertex', 'unique_vertex'),
        'collection': 'User',
        'unique_key': ('user',),
        'fields': {
            'user': 'user'
        },
        'index': [
            {'field' : ('user',), 'unique' : True, 'ttl' : False}
        ]
    },


    # User - rated -> Movie
    # Type 3
    'rated': {
        'type': ('edge', 'unique_edge_on_event'),
        'collection': 'rated',
        'unique_key': (‘date’,),
        '_from_collection': 'User',
        '_from': ('user',),
        '_to_collection': 'Movie',
        '_to': ('movie',),
        'fields': {
            'rating': 'rating',
            'date': 'date'
        },
        'index': []
    },

    # Movie - created_at -> Year
    # Type 2
    ‘created_at’: {
        'type': ('edge', 'unique_edge_btw_vertices'),
        'collection': 'created_at',
        '_from_collection': 'Movie',
        '_from': ('id',),
        '_to_collection': 'Year',
        '_to': ('year',),
        'fields': {
        },
        'index': []
    }
}

스키마가 복잡해보이지만, 한 번 세팅해두면 계속해서 손쉽게 사용할 수 있다.

from pam import client, database, converter

# ArangoDB에 연결
arango_conn = client.get_arango_conn(hosts="http://localhost:8529")
database_obj = database.create_and_get_database(arango_conn, 'movie_ratings', 'root', 'passwd')

# 데이터프레임을 딕셔너리의 리스트 형태로 바꿔준다
data_1 = movies.to_dict('records')
# 전체 스키마를 사용하는 것이 아니기 때문에 사용할 리스트를 지정해준다
data_1_map = ['Movie', 'created_at', 'Year']
converter.arango_converter(data_1, database_obj, schemas, data_1_map)

# Data 2 Upload
data_2 = ratings.to_dict('records')
data_2_map = ['User', 'rated']
converter.arango_converter(data_2, database_obj, schemas, data_2_map)

위 코드만 실행시켜주면 데이터는 성공적으로 들어간다!

5. 다음 단계는

이제 데이터를 모두 넣었다. pam 패키지를 통해 어떠한 데이터 종류든 손쉽게 그래프 형태로 맵핑이 가능하다!

다음 글에서는 실제 들어온 데이터를 가지고 이제껏 배워온 AQL을 뽐내는 시간을 가져보자.

그래프 탐색도 하고, 통계도 뽑아보고, 최종적으로는 추천 시스템을 만들어 볼 예정이다.

아마 다음 글이 아랑고DB 시리즈의 마지막 글이 될 것 같다 😢

애정을 가지고 작성해온 만큼, 유종의 미를 거둬야겠다!! 지금까지 글을 읽으며 따라와주신 분들께 감사하다는 말씀을 드리고 싶다 :)

6. 어디까지왔나

1. 아랑고DB란? 왜 쓰는가?
2. 아랑고DB 세팅하기 on Ubuntu
3. 아랑고DB 쉘로 붙어서 명령어 체험해보기, 실체 파악해보기
4. AQL(Arango Query Lang) 배워보기 1
5. AQL(Arango Query Lang) 배워보기 2 - RETURN / UPDATE
6. AQL(Arango Query Lang) 배워보기 3 - REPLACE / UPSERT / REMOVE
7. 그래프 개념잡기
8. 그래프 횡단하기 Graph Traversal
9. 데이터 모으기 COLLECT / AGGREGATE / MIN_BY, MAX_BY
10. (지금 보고있는 글)프로젝트. 그래프를 통한 영화 추천시스템 만들어보기 1
11. 프로젝트. 그래프를 통한 영화 추천시스템 만들어보기 2 (최종편)

들어가기에 앞서

이 글은 미디엄에 있는 잘 쓰여진 글과 PyPI 튜토리얼을 나도 실제로 따라하면서 한글로 정리하는 형태로 진행될 예정이다.

또한 모듈을 만들 정도면 기본적인 파이썬 지식이나 쉘 커맨드를 알고 있다고 가정하고 세부적인 설명은 생략한다.

본문에서는 이러한 내용도 모두 짚고 넘어가니, 여기서 설명이 부족하다면 꼭 살펴보자.

1. 모듈? 패키지?

파이썬에서 프로그램이 길어지게 되면 스크립트를 여러 파일로 분할하게 된다. 이때, .py 형태로 파일을 나누게 되고, 이들을 각각 모듈 modules 이라고 부른다.

모듈에서 선언된 모든 함수와 선언들은 다른 모듈에서 임포트 될 수 있다.

패키지는 파이썬의 모듈을 구조화하는 형태를 의미하고, 모듈들의 집합을 나타낸다.

우리는 여러 모듈로 구성된 패키지를 만들 예정이다!

심심하면 여기를 더 읽어보자.

2. 패키지 만들어보기

나는 Ubuntu Shell을 통해 해당 작업을 진행할 예정이다. 아마 일부 커맨드를 제외하고는 파이썬 명령어를 쓸 일이 더 많을 것 같으니 OS가 달라도 괜찮지 않을까..?

1) 디렉토리 구조 세팅해주기

PyPI의 튜토리얼을 보면 디렉토리 경로를 설정해주는 것부터가 시작이다. 여기서 사용자가 내 패키지를 어떤 이름으로 임포트 할 지 미리 생각해두면 좋다.

왜냐하면 실제로 패키지를 인스톨하는 pip install 패키지이름 명령어와 import 패키지이름의 명령어에서 쓰이는 이름은 다를 수 있기 때문이다.

만약 내가 A라는 패키지 이름을 원했지만, 이미 PyPI 패키지에 유사한 이름이 있는 경우 이 이름을 사용하지 못한다!

그럼에도 임포트는 A라는 이름으로 하길 원하면 패키지의 디렉토리 이름을 그렇게 설정해주면 된다.

아래는 내가 설정한 디렉토리 구조의 예시이다. 나는 아랑고DB를 쉽게 맵핑해주는 파이썬 패키지를 만들고 있다. 이름은 python-arango-mapper이며, 임포트는 pam으로 하고싶다.

python_arango_mapper/
├──LICENSE
├──README.md
├──setup.py
├──src/
│   └── pam/
│       ├──__init.py
│       ├──그리고 패키지에서 사용하는 다른 파일들

위에서 보면 패키지 루트 디렉토리에는 패키지와 관련된 메타 정보를 갖는 파일들이 포함된다.

• LICENSE는 내 패키지의 이용 권리에 관한 disclaimer이며, 여기에서 나에게 맞는 타입을 골라서, 복사해서 텍스트 그대로 붙여넣으면 된다.
• README.md는 깃 저장소를 최초 만들 때 생성되는 파일과 동일하다. 패키지에 대한 설명을 넣어주면 된다.
• 그다음 가장 중요한 파일이 setup.py인데, 내 파이썬 패키지에 대한 모든 설정 정보들을 포함한다.

2) setup.py 세팅

내가 만든 setup.py는 아래와 같이 구성되어 있다.

from setuptools import find_packages, setup
import setuptools

install_requires = [
    'bleach==4.1.0',
    'certifi==2021.10.8',
    'cffi==1.15.0',
    'charset-normalizer==2.0.9',
    'colorama==0.4.4',
    'cryptography==36.0.1',
    'dataclasses==0.8',
    'docutils==0.18.1',
    'idna==3.3',
    'importlib-metadata==4.8.3',
    'jeepney==0.7.1',
    'keyring==23.4.0',
    'packaging==21.3',
    'pkg_resources==0.0.0',
    'pkginfo==1.8.2',
    'pycparser==2.21',
    'Pygments==2.10.0',
    'PyJWT==2.3.0',
    'pyparsing==3.0.6',
    'python-arango==7.3.0',
    'readme-renderer==32.0',
    'requests==2.26.0',
    'requests-toolbelt==0.9.1',
    'rfc3986==1.5.0',
    'SecretStorage==3.3.1',
    'setuptools-scm==6.3.2',
    'six==1.16.0',
    'tomli==1.2.3',
    'tqdm==4.62.3',
    'twine==3.7.1',
    'typing_extensions==4.0.1',
    'urllib3==1.26.7',
    'webencodings==0.5.1',
    'zipp==3.6.0'
]

with open("README.md", "r", encoding="utf-8") as fh:
    long_description = fh.read()

setup(
    name='python-arango-mapper',
    package_dir={"": "src"},
    packages=setuptools.find_packages(where="src"),
    python_requires=">=3.6",
    install_requires=install_requires,
    version='0.1.5',
    description='fast and easy-to-use python-arango mapper library',
    long_description=long_description,
    long_description_content_type="text/markdown",
    author='lee ui dam',
    author_email='ud803da@gmail.com',
    url='https://github.com/ud803/python-arango-mapper',
    license='GNU General Public License v3.0'
)

• 패키지를 구동하기 위해 필요한 최소한의 패키지 목록을 install_requires에 문자열의 리스트로 기록해준다
• name은 PyPI에 등록할 패키지의 정식 이름이다. 중복되거나 부적절한 경우 등록시 에러가 난다.
• package_dir와 packages는 패키지에 포함할 소스의 위치에 관한 설정이며, 나는 위처럼 해주었다.
• version은 내가 설정하는 내 패키지의 버전을 의미한다. 나도 아직 버저닝에 익숙하지는 않지만, **동일한 버전인 경우 중복 & 수정 업로드가 불가능하다!! **
• long_description은 PyPI 에 올라갈 내 패키지 설명 페이지에 들어갈 내용이다. long을 쓰지 않으면 README.md가 아닌 짧은 설명만 올라가길래 이렇게 설정했다

3) 배포 버전 만들기

이제 준비가 다 끝났으니, 실제 파이썬 명령어를 통해 배포 버전을 만들고, 배포해보자.

# pip 업그레이드
python3 -m pip install --upgrade pip

# build 패키지 업그레이드
python3 -m pip install --upgrade build

# setup.py가 있는 디렉토리에서 실행, 패키지를 빌드한다
python3 -m build

이렇게하면 python_arango_mapper/dist/ 폴더에 두 개의 파일이 생긴다. 이제 업로드해보자.

업로드 전에 테스트 PyPI와 실제 PyPI서 계정만 먼저 만들어주자!

PyPI는 테스트 사이트에 먼저 업로드하는 것을 권장한다.

4) 테스트 업로드하기

# twine 패키지 업그레이드
python3 -m pip install --upgrade twine

python3 -m twine upload --repository testpypi dist/*

테스트 PyPI에 성공적으로 업로드가 되었으면, 계정으로 로그인해서 내 프로젝트가 잘 들어간 것을 확인하면 된다.

5) 실제 업로드하기

테스트까지 끝났으면, 이제 정말로 세상에 내 패키지를 배포해보자.

다 동일하고, 업로드 대상 저장소만 바꿔주면 된다. 그런데 PyPI가 이미 디폴트 저장소이기 때문에 옵션만 지워준다.

# 실제 PyPI에 첫 패키지 배포!!
python3 -m twine upload dist/*

짝짝짝!! 축하한다. 드디어 내 첫 패키지가 세상에 발을 디뎠다.

3. 패키지 테스트해보기

이제 내가 만든 패키지가 잘 동작하는지 바로 확인해보자.

pip install python-arango-mapper를 통해 설치를 하고, 내가 테스트했던 코드들을 실험해보면 된다!

4. 마치며

실제 패키지를 배포할 때는 깃을 통해 버전을 관리하고, 테스트 코드를 작성하는 등의 과정을 거친다.

나는 귀찮아서.. 테스트 코드는 작성하지 않았지만 깃허브에 올려두고 코드 관리를 하고 있다.

혹시 글이 도움이 되셨다면 python-arango-mapper 깃허브에 가서 스타를 꾸욱 눌러주시면 정말 감사하겠다!! 😗