[12주차_팀플03] End-to-end 데이터 파이프라인 구성하기

팀플03. End-to-end 데이터 파이프라인 구성하기
: TMDB 이용한 영화 흥행 및 트렌드 분석

 

---

 
 

프로젝트 기간 : 2024년 1월 8일 (월) ~ 1월 12일 (금) 총 5일
실제 작업 기간 : 2024년 1월 7일 (일) ~ 1월 12일 (금)

 

(대략 20시간 정도 투자)
 

 

 

1. 프로젝트 소개
2. 프로젝트 진행 내용
3. 프로젝트 결과
4. 회고

 

 

 

 

 

프로젝트 소개

프로젝트 공지

본 프로젝트의 목표는 2차 프로젝트 때 개발한 결과물을 바탕으로, 완전한 데이터 파이프라인을 구축하여 데이터가 주기적으로(매일 혹은 매 시간) 되도록 변경하고 Airflow 상의 개발과 운영 경험 및 데이터 파이프라인에서 데이터 웨어하우스(Redshift) 사용 방법을 배우는 것이 목표입니다(필히 이전 결과물을 사용하지는 않으셔도 좋습니다.). 구체적인 기술로는 Airflow, Redshift, Docker 등이 사용되며 본 프로젝트를 통해 CI/CD 구축 방법을 보다 구체적으로 이해할 수 있게 됩니다. 

 

 

 

 

프로젝트 주제 선정

1. 농산물 데이터를 통한 종류별 거래량, 수출입, 수익, 인기 분석
2. 날씨에 따른 실시간 유동인구 분석을 통한 핫플레이스 도출
   • https://data.seoul.go.kr/dataVisual/seoul/guide.do
3. TMDB 영화 데이터를 통한 흥행 요인 및 트렌드 분석
   • https://developer.themoviedb.org/reference/discover-movie
   • https://developer.imdb.com/non-commercial-datasets/
4. Git Archive 활용 랭귀지/스택 분석
   • https://www.gharchive.org
5. 서울시 지하철 데이터 이용
   • https://data.seoul.go.kr/dataList/31/literacyView.do
6. 물류 데이터 이용

다양한 주제 후보가 있었지만, 최종적으로 선정된 것은 TMDB 영화 데이터를 통한 흥행 요인 및 트렌드 분석이다. 하지만 TMDB의 데이터 형태를 파악하여 세부적인 주제는 다시 정하기로 하였다. 1, 2번이 나의 아이디어였고 1번은 데이터를 어디서 찾아오면 좋을지 못 찾았다. 4번 아이디어가 좋은 반응이었지만 데이터 용량이 너무 커서 과금없이 해결하기에는 무리가 있을 것 같아서 무난하고 하기 쉬울 것 같은 영화 데이터를 선택했다.

 

 

 

 

---

 
 
 

프로젝트 진행 내용 (TIL)

맡은 역할 : 상품 상세 정보 및 고객 관련 데이터 크롤링, 테이블 모델링, 차트 생성 및 대시보드 구성, 노션에 프로젝트 정리하기

 

 

 

사전 미팅

일요일 오후 2시, 첫 미팅을 가졌다. 각자 생각해온 주제를 논의했고, 실현 가능성을 파악하여 다시 월요일에 주제를 확정하기로 하였다.

 

실시간 유동 인구 파악으로 인한 “핫플” 분석
크리스마스 등 어디를 놀러가면 좋을지
날씨에 따른 핫플 ?

농산품 종류별 거래량 비교하여 인기 분석 
(얼마나 생산되는지, 얼마나 팔리고, 수익은 어느정도이며, 수입수출은 어느정도인지)
+ 가격 등 정보
But, API를 아직 찾지 못하였다.

영화 데이터를 이용하여 흥행 요인, 트렌드 분석 1
TMDB API 사용하기 -> 데이터 잘 정리되어 있고 하루 호출 제한 없음
영화 목록, 장르 정보, 평균/평점 값

서울시 지하철 데이터 1
일별, 시간별 >> 데이터 모아서 보여줄 때는 좋다
일일 승하차도 하루에 한 번씩 업데이트 해준다
서울시 데이터포털에서 API로 받아오기
+ 기상청 특보 자료 : 호우 주의보, 대설 주의보에 따라 지하철 이용량 분석

GH Archive 이용 랭귀지/스택 분석 3
깃허브 API를 통해 올린 덤프로 분석하기
1시간마다 업데이트됨 (데이터 크기는 보통 250MB ~ 800MB)
클라우드 스토리지 최대 5GB 
용량이 커서 최대 1일치? (아슬아슬할수도)
최신 1시간 or 최신 24시간 트렌드
오픈소스, 커맨드 등 분석

물류 데이터 프로젝트
API 대신 자동 스크래핑으로 이용

 

 

---

 

1일차

주제 선정

 
TMDB를 이용한 영화 분석을 주제로 결정했다. 어떤 툴을 이용할까 논의하다가 우선 한 명이 빅쿼리에 데이터를 옮겨놓기로 하였다. 샘플 데이터를 먼저 올려놓고 공유해주면 어떤 데이터를 가져와서 시각화할 것인지는 각자 생각해보기로 하였다.

 

 

사용 API : Movie

• 하루 호출 제한이 없음
• 초당 50번의 요청 가능

데이터 수집

• 영화 데이터 수집 : tmdb API 사용
  • tmdb api에서 영화 목록을 제공하는 discover api 사용
  • with_origin_country 옵션을 통해 한국 영화만 수집 가능
  • 장르 정보, 평가 점수
  • 매일 변하는 값 : popularity, vote_count

• 영화 세부 정보
  • 수익, 예산, 러닝타임 (수익이 0으로 측정되는 영화도 있음(옛날 영화의 경우))
  • 장르를 아이디, 장르명을 같이 제공
  • 세부정보
• 키워드

 

대시보드 구성

• 평점이 높은 장르
• 평점이 높은 영화와 수익의 연관성
• 영화 제작 국가별로 인기 있는 장르
• 예산과 수익의 연관성
• 개봉일을 기준으로 선호되는 장르

 

 

아래와 같이 세팅되었다. 한번에 가져올 수 있는 페이지 제한이 있어서 우선은 10000개만 샘플로 넣어져있다.

• 프로젝트ID : tmdb-movies-410603
• 서비스 key (Bigquery, GCS) : 첨부파일

tmdb-movies-410603-bd2283f086a0.json

0.00MB

 

• 참고 문서 : DBeaver 빅쿼리 접근 가이드 https://velog.io/@2innnnn0/dbeaverguide

 

 

 

---

 
 

2일차

샘플 데이터 연결 후 데이터 확인
테이블 모델링
preset.io (superset) 대시보드 생성

 

 

1. 샘플 데이터 연결 후 데이터 확인

DBeaver 통하여 아래와 같이 샘플 데이터를 연결하였다. 이 샘플 데이터를 바탕으로 상세 주제와 테이블 모델링, 어떤 분석을 할 것인지에 대해 고민하고 테스트해보면 된다.

 

• vote_count
• vote_average
• status
• original_language
• release_date
• video
• original_title
• production_companies
• spoken_languages
• homepage
• title
• belongs_to_collection
• budget
• overview
• genres
• poster_path
• imdb_id
• id
• tagline
• runtime
• revenue
• production_countries
• popularity
• adult
• backdrop_path

 

나의 아이디어는 총 3가지로 구분하여 데이터를 분석하는 것이다.

1. 명예의 전당 : 장르 분석 혹은 영화 분석 내용
   • 평균 평점이 높은, 최고 예산, 최고 흥행, 최고 수익
   • 영화 제작 1위, 어떤 언어가 가장 많이 제작되었는지는 당연히 미국이기 때문에 무의미할듯
2. 장르 트렌드 : 어떤 장르가 많이 제작되고, 어떤 장르가 인기 있는지
3. 흥행 분석 : budget, revenue, popularity를 이용하여
   • 예산 대비 수익의 분포 분석, 예산 및 수익의 연도별 흐름 분석 + 장르별 매출액 분석?
   • 평균 수익 및 평균 예산 분석, 예산 대비 수익이 높은 영화, 예산 대비 수익이 가장 높은 영화 너무 흔하다

우려사항 : 

• 한 영화에 장르가 여러 개 있어서 처리하는 데에 어려움이 있다. -> 일단 파이썬으로 처리 후 SQL 작업
• 이미지 처리를 할 수 있는가

 

 

1) 명예의 전당

# 1. vote_average를 사용하여 가장 평점이 높은 영화의 id, 평점, original_title, title, poster_path
SELECT id, vote_average, vote_count, original_title, title, poster_path
FROM dev.movie_samples
WHERE vote_count > 1000
ORDER BY vote_average DESC
LIMIT 1;

# 4. budget이 가장 높은 데이터 하나
SELECT id, budget, original_title, title, poster_path
FROM dev.movie_samples
ORDER BY budget DESC
LIMIT 1;

# 5. popularity가 가장 높은 데이터 하나
SELECT id, popularity, original_title, title, poster_path
FROM dev.movie_samples
ORDER BY popularity DESC
LIMIT 1;

# 6. revenue가 가장 높은 데이터 하나
SELECT id, revenue, original_title, title, poster_path
FROM dev.movie_samples
ORDER BY revenue DESC
LIMIT 1;

 

# 2. production_countries을 분석하여 가장 많은 빈도의 데이터 하나
SELECT production_countries, COUNT(*) AS frequency
FROM dev.movie_samples
GROUP BY production_companies
ORDER BY frequency DESC
LIMIT 1;

# 3. spoken_languages을 분석하여 가장 많은 빈도의 데이터
SELECT spoken_languages, COUNT(*) AS frequency
FROM dev.movie_samples
GROUP BY spoken_languages
ORDER BY frequency DESC
LIMIT 1;

import pandas as pd
import json

# 파일 경로 설정
file_path = 'sample_data.csv'

# 파일 읽어오기
df = pd.read_csv(file_path)

# production_countries의 상위 3개 빈도 계산
top_production_countries = df['production_countries'].value_counts().nlargest(3)

# spoken_languages의 상위 3개 빈도 계산
top_spoken_languages = df['spoken_languages'].value_counts().nlargest(3)

# 상위 3개 빈도 출력
print("Top 3 production_countries:")
print(top_production_countries)

print("\nTop 3 spoken_languages:")
print(top_spoken_languages)

 

 

2) 장르 트렌드 분석

# 1) 장르별 영화
import pandas as pd
import json
import plotly.express as px

# 파일 경로 설정
file_path = 'sample_data.csv'

# 파일 읽어오기
df = pd.read_csv(file_path)

# 각 장르별 영화 갯수를 저장할 딕셔너리 초기화
genre_movie_counts = {}

# 각 행에서 장르 정보 추출
for index, row in df.iterrows():
    genres_info = json.loads(row['genres'])['v']

    for genre_info in genres_info:
        genre_name = genre_info['v']['f'][0]['v']

        # 각 장르별 영화 갯수 계산
        genre_movie_counts[genre_name] = genre_movie_counts.get(genre_name, 0) + 1

# 영화 갯수 데이터프레임 생성
movie_count_df = pd.DataFrame({
    'genres': list(genre_movie_counts.keys()),
    'movie_count': list(genre_movie_counts.values())
})

# 트리맵 그리기
fig = px.treemap(movie_count_df, 
                 path=['genres'], 
                 values='movie_count',
                 title='Movie Count Distribution by Genre',
                 color='movie_count',
                 color_continuous_scale='Viridis')

fig.show()


# 2) 장르별 관객 수
import plotly.express as px

# 수평 막대 그래프 그리기
fig = px.bar(totals_df, 
             x='popularity_total', 
             y='genres', 
             orientation='h',
             title='Genre Popularity Distribution',
             labels={'popularity_total': 'Total Popularity', 'genres': 'Genres'},
             )

fig.show()



# totals_df를 사용하여 트리맵 그리기
fig = px.treemap(totals_df, 
                 path=['genres'], 
                 values='popularity_total',
                 title='Genre Popularity Distribution',
                 color='popularity_total',
                 color_continuous_scale='Viridis')

fig.show()

 

 

 

3) 흥행 분석

# 1) 예산에 따른 평점
import plotly.express as px

# 예산과 평점이 0 또는 null인 영화는 제외
filtered_df = df[(df['budget'] > 0) & (df['vote_average'] > 0)]

fig = px.scatter(filtered_df, 
                 x='budget', 
                 y='vote_average',
                 title='Budget vs. Vote Average',
                 labels={'budget': 'Budget', 'vote_average': 'Vote Average'},
                 hover_data=['original_title'])

fig.show()


# 2) 예산, 수익, 관객수 분석
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

# CSV 파일에서 데이터프레임으로 읽어오기
df = pd.read_csv('sample_data.csv')

# 예산, 수익, 인기도가 모두 양수인 데이터 선택
df = df[(df['budget'] > 0) & (df['revenue'] > 0) & (df['popularity'] > 0)]

# 바이올린 플롯
plt.figure(figsize=(12, 6))
sns.violinplot(x='variable', y='value', data=pd.melt(df[['budget', 'revenue', 'popularity']]))
plt.title('Violin Plot of Budget, Revenue, and Popularity')
plt.show()

# 산점도 행렬
sns.pairplot(df[['budget', 'revenue', 'popularity']])
plt.suptitle('Scatter Plot Matrix of Budget, Revenue, and Popularity', y=1.02)
plt.show()

 

 

미팅에서 각자 대시보드를 구성해보기로 하여 몇 가지 차트만 선정하여 테이블 모델링 후 dbt를 수행해볼 예정이다.

<회의록>
대시보드
- Apache Superset으로 결정됨
- Superset App을 통한 환경구성(1안) 
- https://www.supersetapp.com/각자 local 환경에서 작업후 chart 및 dashboard yml 파일 공유(2안)
R&R 분담
각자 만들고자 하는 차트 기준으로 E2E 파이프 라인 전 과정에 대해 각자 역할 분담
개인별 담당 역할 예시 : 데이터 탐색 및 차트 구성, 차트 생성에 필요한 테이블 모델 정의(중간단계 테이블 포함, dbt), dbt 스케쥴링을 위한 airflow dag 작성
프로젝트 GCP 및 github 구성
GCP 프로젝트 및 github organization 초대를 위한 계정 정보 공유 요청

 

 

 

 

2. 테이블 모델링

1) 명예의 전당

1. 평균 평점 테이블
2. 최다관객/최대예산/최고수익 테이블

• 평균 평점 : id (PK), vote_average, vote_count, original_title, title, poster_path, production_countries
  • WHERE vote_count > 1000 으로, 유의미한 평점 도출
• 최고 인기점수 : id (PK), popularity, original_title, title, poster_path, production_countries
• 최대 예산 : id (PK), budget, original_title, title, poster_path, production_countries
• 최고 수익 : id (PK), revenue, original_title, title, poster_path, production_countries

 

2) 장르 트렌드 : 어떤 장르가 많이 제작되고, 어떤 장르가 인기 있는지

• 파이썬을 통해 data 변환을 해야 하므로 추후 확인 가능 (ETL 과정에서 변환을 해야 한다)

 

3) 흥행 분석

• id, title, budget, revenue, popularity, release_date

1. 예산 및 수익, 흥행의 연도별 흐름 분석
   • EXTRACT(YEAR FROM release_date), SUM(popularity), SUM(budget), SUM(revenue)
2. 예산 대비 수익이 높은 영화 TOP 100
   • id, title, budget, revenue, release_date
3. 예산, 수익, 흥행 산점도

# 예산 및 수익, 흥행의 연도별 흐름 분석
SELECT
    EXTRACT(YEAR FROM release_date) AS release_year,
    SUM(budget) AS total_budget,
    SUM(revenue) AS total_revenue,
    SUM(popularity) AS total_popularity
FROM
    dev.movie_samples
WHERE
    budget > 0 AND revenue > 0
GROUP BY
    release_year
ORDER BY
    release_year;

# 예산 대비 수익이 높은 영화 TOP 100
SELECT
    id,
    title,
    budget,
    revenue,
    revenue / budget AS revenue_to_budget_ratio,
    release_date
FROM
    dev.movie_samples
WHERE
    budget > 0 AND revenue > 0
ORDER BY
    revenue_to_budget_ratio DESC
LIMIT 100;

 

 

최종적인 테이블 모델링 : 

1. 기본 정보 테이블 : id (PK), original_title, title, poster_path, production_countries
2. 수치 정보 테이블 : id (PK), original_title, title,  popularity, budget, revenue, release_date

-- 기본 정보 테이블 생성
CREATE TABLE dev.movie_basic_info AS
SELECT
    id,
    original_title,
    title,
    poster_path,
    production_countries
FROM
    dev.movie_details;

-- 수치 정보 테이블 생성
CREATE TABLE dev.movie_numeric_info AS
SELECT
    id,
    original_title,
    title,
    popularity,
    budget,
    revenue,
    release_date
FROM
    dev.movie_details;

 

일단 테이블 모델링을 하였으니, 이제 다음으로 테이블 생성하여 preset에서 대시보드 만들어보고, DBT랑 DAG 작성을 해보면 될 것 같다.

 

 

 

3. preset.io (superset) 대시보드 생성

preset에서 무료로 이용 가능하다. 무료 기능은 아래와 같다.

• 1 workspace
• 5 team members
• All team members must be Team Admins
• All team members must be Workspace Admins
• No advanced permissions using Data Access Roles or Row Level Security
• No reports or alerts on charts and dashboards
• Datasets created from dbt models will no longer sync with dbt Cloud

 

1) Database 연결 : settings에서 빅쿼리 DATABASE를 연결하였다.

2) Dataset 생성

3) Chart 제작

4) Dash board 연결 및 제작

 

 

 

 

---

 

 

3일차

preset.io (superset) 차트/대시보드 생성
dbt 학습

 

1) Database 연결 : settings에서 빅쿼리 DATABASE를 연결하였다.

2) Dataset 생성 : 미리 만들어둔 데이터셋을 연결했다.

3) Chart 제작 : 

• 최근 1년간 분기별 영화 제작 → 연간 영화 제작 5년간 추이
• 연간 등록 영화
• 연도별 예산 및 수익의 흐름
• 예산 대비 수익 (산점도)
• 예산 대비 평점
• 예산 대비 수익이 높은 영화 TOP 10 → 고수익 영화 TOP 10의 예산 비교, 고예산 TOP 10의 수익 비교

추가로 데이터가 필요해서 빅쿼리 내에서 열을 추가했다. 해당 내용을 기존에 추가한 데이터셋에 반영하려면 Edit dataset에서 Sync해주어야 한다. 하지만 예산 대비 수익을 구해도 수익이나 예산이 제대로 작성되지 않은 항목이 있어서 수익이나 예산을 토대로 차트를 그려야 했다.

-- 기존 테이블에 열 추가
ALTER TABLE dev.movie_numeric_info
ADD COLUMN revenue_to_budget_ratio FLOAT64;

-- 기존 데이터에 값을 업데이트
MERGE dev.movie_numeric_info AS target
USING (
    SELECT
        id,
        revenue / budget AS revenue_to_budget_ratio
    FROM
        dev.movie_numeric_info
    WHERE
        budget > 0 AND revenue > 0
) AS source
ON
    target.id = source.id
WHEN MATCHED THEN
    UPDATE SET
        target.revenue_to_budget_ratio = source.revenue_to_budget_ratio;

 

 

• 영화 전체 평균 평점
• 투표를 통한 영화 관심도

깜빡하고 평점을 가져오지 않아서 빅쿼리에서 먼저 추가해주었다.

ALTER TABLE dev.movie_numeric_info
ADD COLUMN vote_average FLOAT64,
ADD COLUMN vote_count INT64;

MERGE INTO dev.movie_numeric_info numeric
USING dev.movie_details details
ON numeric.id = details.id
WHEN MATCHED THEN
  UPDATE SET
    numeric.vote_average = details.vote_average,
    numeric.vote_count = details.vote_count;

 

 

4) 대시보드 연결 및 생성

 

에러사항 : 

• genre가 id, name으로 구분되어 나오긴 하지만 실제 사용하기에 에러가 발생한다. (대시보드 툴 사용 미흡일 수도)
• budget, revenue가 0이거나 IMDB와는 다르게 유난히 낮게 작성되어 있는 수치들이 많아서 이를 중점으로 한 분석은 무의미해진다. 

 

이렇게 대시보드 1차 구상하였고, 다음으로 dbt를 살펴볼 예정이다. 대시보드는 추후에 수정/보완이 필요하다.

 

 

 

dbt 학습

https://www.youtube.com/watch?v=jvy3aoy462k

안녕하세요 여러분, 이 아침에 함께해주셔서 감사합니다. 저는 여기 preset의 클라우드와 dbt 코어의 통합에 대해 조금 이야기하기 위해 Beto와 함께 있습니다. 제 이름은 Sharisham이고, 저는 preset에서 개발 관련 엔지니어입니다. 이제 Beto에게 자신을 소개하도록 할게요. 안녕하세요 여러분, 제 이름은 Beto이고, 저는 preset에서 스태프 엔지니어로 일하고 있습니다. 나는 지금까지 superset과 함께 다섯 년 동안 일해왔어. 여러분과 함께할 수 있어 기쁘다.

좋아, 그럼 먼저 preset이 무엇인지에 대해 조금 얘기해보겠습니다. superset이 무엇인지에 대해 먼저 얘기하겠다. superset은 현대적이고 오픈 소스 BI 플랫폼으로 거의 모든 SQL을 사용하는 데이터 엔진과 작동합니다. 이 플랫폼은 데이터 시각화를 위한 다양한 차트를 제공하며 대시보드를 생성할 수 있습니다. 일반적으로 superset 내에서는 두 가지 주요 워크플로우가 있습니다. 첫 번째는 "탐색"이라고 하는데, 이는 노코드 차트 빌더 인터페이스를 제공하여 SQL Alchemy를 통해 사용자를 대신하여 쿼리를 생성합니다. 이 워크플로우를 사용하는 주체는 주로 데이터 탐색가나 데이터 소비자, 또는 데이터에 대한 통찰을 얻으려는 일반 사용자입니다. 여기서는 SQL을 직접 작성하지 않고도 데이터에 대한 통찰을 얻을 수 있습니다. 다른 워크플로우는 SQL Lab을 통해 지원되며, 여기서는 완전한 SQL IDE를 제공합니다. 이 워크플로우는 주로 데이터 엔지니어, 데이터 과학자 또는 기술적인 데이터 사용자 및 탐색가에 의해 사용됩니다. 그리고 preset은 다중 워크스페이스 클라우드 호스팅 superset으로, superset의 모든 기능을 제공하며 엔드 유저 문서 사용을 위한 소프트 컴플라이언스를 레이어로 추가하며 행 기반 액세스 제어, 싱글 사인온 및 엔터프라이즈 지원과 같은 특정 제어도 제공합니다. 또한 현재 베타 상태인 임베디드 SDK에 대한 액세스도 제공합니다.

이제 dbt는 데이터 변환 도구로서, 팀이 소프트웨어 엔지니어링 모범 사례를 따라 분석 코드를 빠르게 협업하여 배포할 수 있게 해줍니다. 모듈성, 이식성, CI/CD 및 문서화와 같은 소프트웨어 엔지니어링의 최선의 방법을 따릅니다. 이를 통해 SQL을 알고 있는 사람들이 생산 품질의 데이터 파이프라인을 쉽게 구축할 수 있게 됩니다. dbt는 모든 데이터 모델을 버전 관리할 수 있도록 허용하며, 개발 환경을 안전하게 배포할 수 있게 해줍니다. Git을 지원하는 버전 관리는 협업 및 필요에 따라 이전 상태로 쉽게 복귀할 수 있도록 합니다. 또한, 프로덕션 이전에 각 모델을 테스트하고 문서화할 수 있으며, dbt는 동적으로 문서를 생성하여 모든 데이터 이해자와 공유할 수 있도록 해줍니다. 마지막으로 dbt를 사용하면 SQL로 모든 데이터 모델을 손쉽게 개발할 수 있습니다. 이는 SELECT 문을 사용하여 모듈식 SQL 모델을 작성하고, 동일한 프로젝트 내에서 다른 dbt 모델을 참조할 수 있는 dbt의 참조 함수를 사용하여 의존성 관리를 dbt 자체에게 맡기게 됩니다.

다음으로 Beto가 통합 자체에 대해 더 자세히 설명할 것입니다.

이번 세션에서는 Superset과 dbt의 통합을 다루었습니다. 이는 데이터 엔지니어링 및 BI 도구 간의 협업을 용이하게 하고, 데이터 모델의 신뢰성을 확보하는 데 도움이 됩니다.

dbt Integration:
dbt (Data Build Tool)를 사용하여 데이터 모델을 정의하고 생성합니다. 
이는 데이터 엔지니어링에서 데이터를 변환하고 가공하는 데 도움이 되는 강력한 도구입니다.
dbt CLI를 통해 데이터 모델을 빌드하고, 데이터베이스 및 스키마를 정의합니다.

Superset Integration:
Superset은 BI 도구로, 시각화 및 대시보드를 만드는 데 사용됩니다.
Superset과 dbt 간의 통합을 통해 dbt에서 정의한 데이터 모델을 쉽게 시각화할 수 있습니다.

Syncing Assets with Superset CLI:
Superset CLI를 사용하여 dbt에서 정의한 데이터 모델, 대시보드, 차트 등의 자산을 Superset으로 동기화합니다.
Superset CLI는 YAML 파일을 읽어서 Superset 워크스페이스에 자산을 배포합니다.

Asset Management with Source Control:
dbt에서 정의한 데이터 모델은 source control에 저장됩니다.
Superset에서 생성한 대시보드 및 차트도 source control에 추가하여 버전 관리할 수 있습니다.

Automated Documentation:
dbt의 자동 생성된 문서를 Superset에서 활용합니다. 
데이터 모델의 lineage 및 설명이 자동으로 Superset에서 표시됩니다.

Read-Only Dashboards and Charts:
생성된 대시보드 및 차트를 수정하지 못하도록 설정할 수 있습니다. 
이를 통해 데이터의 신뢰성을 유지할 수 있습니다.

Future Plans:
Github Actions을 사용하여 Superset 및 dbt 자산을 자동으로 업데이트하는 방법을 소개했으며, 앞으로 더 많은 업데이트와 향상이 예정되어 있습니다.

이러한 통합은 데이터 엔지니어 및 BI 전문가들이 데이터 모델을 쉽게 관리하고, 시각화된 결과물을 신뢰성 있게 생성하는 데 도움이 됩니다.

 

https://docs.preset.io/docs/the-preset-cli

https://docs.preset.io/docs/dbt-sync

dbt Integration

 

일단 preset 대시보드 만들었고, dbt 동기화 방법은 찾았다.

 

그럼 dbt로는 무엇을 작업하면 되는지?? 아예 감이 안 잡힌다.

1) API 통해서 전체 데이터를 Airflow DAG로 가져오면 (raw data)

2) dbt로 이 raw data를 별도의 분석용 View로 만드는 작업을 해주는 것인가? (Input)

- 이 때 yml 파일 만들어지나??

3) 그리고 분석용 View 테이블에서 SQL구문을 작성하면 되나? (Output)

- 근데 이 때, preset은 그냥 정해진 차트에 따라서 컬럼만 선택해주는 등 작업을 했는데

- dbt로는 어떻게 해야 하는 거지?

 

멘토님 미팅 : 

• Airflow 권장 방식 : 단순히 트리거 용도로만
• 클라우드 펑션
• Docker로 파이프라인 같이 구축하기
• 데이터 마트는 어떤 것으로?
• dbt로 데이터 마트 그리기

 

DAG 생각해보기

• ELT DAG : TMDB에서 API로 데이터 추출하고, id Uniqueness 보장하도록 전처리 후에 빅쿼리에 raw 데이터 적재
  • extract : TMDB에서 API로 데이터 추출하기 (Incremental update)
  • transform : 데이터 변환 필요한 부분 처리하기
    • id Uniqueness 보장하도록
  • load : 원래는 Data lake에 raw data로 적재해야 하는데, 일단 DW인 빅쿼리에 raw data로 적재하기
• ELT DAG : 빅쿼리에서 raw -> analysis 테이블로 복제 및 변환
  • extract : 빅쿼리에서 데이터 추출하기 raw -> staging = dev -> analysis
  • load : 분석용 테이블 가져오기 staging -> core 
  • transform : 개별 분석용 데이터 sql로 변환
• TEST DAG
  • PK Uniqueness 검증하기 : id에 중복이 없는지? NOT NULL
  • ETL, ELT 잘 작동되는지

 

테스트는 보통 언제 처리하는지?

분석용 테이블이 대시보드 툴에도 주기적으로 반영되게 하려면 어떻게 해야 하는지? dbt로 데이터마트?

dbt와 airflow 트리거 역할 분담은 어떻게 하면 되는지?

 

일단 오퍼레이터 무엇을 사용할 수 있는지 찾아보고,

dbt 시도해보고, DAG 구현해보고

 

---

 

4일차

Airflow DAG
- ELT DAG

 

 

필요한 DAG

• TMDB에서 데이터 수집하여 raw_data로 적재하는 DAG
• raw -> analysis 변환하는 DAG (두 테이블 모두 빅쿼리 내에서 진행, 스키마 별도로 할지?)
  
  • dbt 이용 가능하다면 yml 파일 생성 필요
  • dbt 어려우면 그냥 빅쿼리에서 변환하는 airflow DAG로 진행
• 테스트 DAG 

 

팀원 3명이 대답이 없어서 혼자라도 하는 방향으로 위와 같이 DAG를 설정했다.

 

크게 ETL, ELT DAG를 각각 만드는 것인데, 일단 ELT DAG부터 하려고 한다.

 

---

 

ELT DAG

1) 빅쿼리에서 analysis 스키마를 만들었다.

CREATE SCHEMA analysis;

 

근데 그냥 dev 스키마 내에서 하기로 결정했다. 이미 차트에 연결시킨 것도 dev 스키마 내이고, 테이블의 수가 많지 않기 때문이다. 스키마는 별도로 분리하는 것이 좋다는 것을 기억해두고 다음부터 반영하기로 했다. 이번에는 굳이 대공사를 할 필요는 없을 것 같다. (preset 렉이 너무 많이 걸려서 시간이 부족하므로 대공사라 칭하였다.)

 

 

2) ELT DAG 코드 생성

새로 추가되는 영화 데이터만이 아니라 기존 영화의 평점도 변할 수 있으므로 일단 Full Refresh 방식으로 가져오기로 하였다. 여건이 된다면 고정적으로 변하지 않는 정보와 평점이나 투표수처럼 변하는 정보를 나누어서 Incremental Update 하는 방식도 고려해보고 싶다.

 

SQL 쿼리는 MERGE INTO 구문을 사용하여 두 테이블을 비교하고, WHEN MATCHED일 때는 업데이트를 수행하고, WHEN NOT MATCHED일 때는 새로운 행을 추가한다. MERGE INTO 쿼리는 id를 기준으로 매칭을 수행한다.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.google.transfers.bigquery import BigQueryOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 14),
    'end_date': datetime(2024, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'update_analysis_tables',
    dag_id = 'update_analysis_tables',
    default_args=default_args,
    description='DAG to update dev. basic, numeric tables',
    schedule_interval='@daily',
)

update_movie_basic_info_task = BigQueryOperator(
    task_id='update_movie_basic_info',
    sql='''
    MERGE INTO dev.movie_basic_info AS dest
    USING dev.movie_details AS src
    ON dest.id = src.id
    WHEN MATCHED THEN
      UPDATE SET
        dest.original_title = src.original_title,
        dest.title = src.title,
        dest.poster_path = src.poster_path,
        dest.production_countries = src.production_countries
    WHEN NOT MATCHED THEN
      INSERT (id, original_title, title, poster_path, production_countries)
      VALUES (src.id, src.original_title, src.title, src.poster_path, src.production_countries);
    ''',
    use_legacy_sql=False,
    bigquery_conn_id='tmdb-movies-410603',
    dag=dag,
)

update_movie_numeric_info_task = BigQueryOperator(
    task_id='update_movie_numeric_info',
    sql='''
    MERGE INTO dev.movie_numeric_info AS dest
    USING dev.movie_details AS src
    ON dest.id = src.id
    WHEN MATCHED THEN
      UPDATE SET
        dest.title = src.title,
        dest.popularity = src.popularity,
        dest.budget = src.budget,
        dest.revenue = src.revenue,
        dest.release_date = src.release_date,
        dest.vote_average = src.vote_average,
        dest.vote_count = src.vote_count
    WHEN NOT MATCHED THEN
      INSERT (id, title, popularity, budget, revenue, release_date, vote_average, vote_count)
      VALUES (src.id, src.title, src.popularity, src.budget, src.revenue, src.release_date, src.vote_average, src.vote_count);
    ''',
    use_legacy_sql=False,
    bigquery_conn_id='tmdb-movies-410603',
    dag=dag,
)

update_movie_basic_info_task
update_movie_numeric_info_task

 

 

3) Airflow DAG 실행 확인

GCP에 서버를 생성하여 Airflow를 설치한 적이 있어서, 여기서 실행해보려고 한다. '브라우저에서 SSH를 통해 연결'하여 접속했다.

cd ~/airflow/dags   # dags 폴더에 들어가서
nano elt_dag.py     # ELT 전용 DAG 파일을 생성하고
# python 코드를 작성해서 control+X -> Y 통해서 저장했다.

 

그 다음에 웹서버를 연결했는데 DAG가 올라오지 않아서... 고생 좀 했다.. Airflow 웹서버 재시작을 하려고 하는데 무언가 에러가 발생한다.

  File "/usr/lib/python3/dist-packages/OpenSSL/crypto.py", line 1573, in X509StoreFlags
    CB_ISSUER_CHECK = _lib.X509_V_FLAG_CB_ISSUER_CHECK
AttributeError: module 'lib' has no attribute 'X509_V_FLAG_CB_ISSUER_CHECK'
[2024-01-11 13:41:55 +0000] [49050] [ERROR] Connection in use: ('0.0.0.0', 8080)
[2024-01-11 13:41:55 +0000] [49050] [ERROR] Retrying in 1 second.
[2024-01-11 13:41:56 +0000] [49050] [ERROR] Connection in use: ('0.0.0.0', 8080)
[2024-01-11 13:41:56 +0000] [49050] [ERROR] Retrying in 1 second.
[2024-01-11 13:41:57 +0000] [49050] [ERROR] Connection in use: ('0.0.0.0', 8080)
[2024-01-11 13:41:57 +0000] [49050] [ERROR] Retrying in 1 second.
[2024-01-11 13:41:58 +0000] [49050] [ERROR] Connection in use: ('0.0.0.0', 8080)
[2024-01-11 13:41:58 +0000] [49050] [ERROR] Retrying in 1 second.
[2024-01-11 13:41:59 +0000] [49050] [ERROR] Connection in use: ('0.0.0.0', 8080)
[2024-01-11 13:41:59 +0000] [49050] [ERROR] Retrying in 1 second.
[2024-01-11 13:42:00 +0000] [49050] [ERROR] Can't connect to ('0.0.0.0', 8080)
[2024-01-11 13:43:50,146] {webserver_command.py:222} ERROR - No response from gunicorn master within 120 seconds
[2024-01-11 13:43:50,147] {webserver_command.py:223} ERROR - Shutting down webserver

 

여기서 OpenSSL 라이브러리와 관련된 _lib.X509_V_FLAG_CB_ISSUER_CHECK라는 속성을 찾을 수 없다는 내용이 포함되어 있다. 이것이 OpenSSL 라이브러리 관련 문제라고 예측할 수 있다. OpenSSL 라이브러리를 업데이트하여 문제를 해결하려고 시도하려고 한다.

• 시도 1. OpenSSL 업데이트 후 웹서버를 다시 시작해보자.

sudo apt-get update
sudo apt-get install --only-upgrade openssl

airflow webserver --port 8080

 

• 시도 2. 파이썬 가상환경에서 업데이트하기

pip install --upgrade openssl

 

• 시도 3. Airflow 환경변수 셋업 실행하기

AIRFLOW_HOME=/var/lib/airflow airflow db init

 

시도3을 하니까 됐다!!! Airflow로 로그인 할 때 (사용자가 바뀔 때)마다 필요한 Airflow 환경변수 셋업이라는 메모를 보고 실행했더니 드디어 연동되었다. 다만 내가 작성한 DAG에서는 모듈 에러가 나서 추가로 설치가 필요했다.

Broken DAG: [/var/lib/airflow/dags/elt_dag.py] Traceback (most recent call last):
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/var/lib/airflow/dags/elt_dag.py", line 3, in <module>
    from airflow.providers.google.transfers.bigquery import BigQueryOperator
ModuleNotFoundError: No module named 'airflow.providers.google'

 

그래서 이렇게 했는데 또 에러났다.

pip install apache-airflow-providers-google

ERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.
launchpadlib 1.10.13 requires testresources, which is not installed.
mysql-connector-python 8.0.32 requires protobuf<=3.20.3,>=3.11.0, but you have protobuf 4.25.2 which is incompatible.

 

 

 

 

---

 5일차

ELT DAG 수정

 

어제 하다가 만 ELT DAG 수정을 마저 해보자.

 

1) PATH에 디렉토리 추가하여

export PATH=$PATH:/home/jbyoon9/.local/bin

 

2) 업그레이드 해주고

python3 -m pip install --upgrade pip

 

3) API 라이브러리를 설치했다.

pip install 'apache-airflow[google]'

https://airflow.apache.org/docs/apache-airflow-providers-google/stable/operators/cloud/bigquery.html

Google Cloud BigQuery Operators — apache-airflow-providers-google Documentation

 

4) apache-airflow-providers-google 설치를 다시 시도하고

pip install apache-airflow-providers-google

 

5) Airflow 관련 환경 설정을 해주었다.

airflow@ vi airflow.cfg   # 여기서 빅쿼리를 연결했는데, 데이터셋ID는 에러가 나서 없이 했더니 됐다.
sql_alchemy_conn = bigquery://tmdb-movies-410603

 

6) 권한이 없어서 권한도 부여해주었다. 

# 1) Ubuntu 계정에서 airflow 폴더 아래에 keys 폴더를 생성했다.
mkdir keys

# 2) key에 대한 json을 만들고 (파일 이동하면 되는데 브라우저로 열었으니까)
nano movies_key.json

# 필요한 경우, key.json 파일 권한 설정
chmod 600 ~/movies_key.json

# 3) 환경변수를 설정한다.
export AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT='{"conn_type": "google_cloud_platform", "extra": {"key_path": “~/airflow/keys/movies_key.json", "scope": "https://www.googleapis.com/auth/cloud-platform", "project": "airflow", "num_retries": 5}}'

https://airflow.apache.org/docs/apache-airflow-providers-google/stable/connections/gcp.html#authenticating-to-google-cloud

Google Cloud Connection — apache-airflow-providers-google Documentation

 

아래 내용을 json에 추가했다.

  "scopes": [
    "https://www.googleapis.com/auth/cloud-platform",
    "https://www.googleapis.com/auth/bigquery"
  ]

 

7) 아무래도 BigQueryOperator는 더이상 존재하지 않아서 에러가 발생하는 듯 했다. 다른 오퍼레이터로 바꿔야한다.

 

 

 

BigQueryUpsertTableOperator

https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/cloud/operators/bigquery/index.html#airflow.providers.google.cloud.operators.bigquery.BigQueryUpsertTableOperator

airflow.providers.google.cloud.operators.bigquery — apache-airflow-providers-google Documentation

 

from datetime import datetime, timedelta
from airflow import DAG
from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryUpsertTableOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 14),
    'end_date': datetime(2024, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'update_analysis_tables',
    dag_id = 'update_analysis_tables',
    default_args=default_args,
    description='DAG to update dev. basic, numeric tables',
    schedule_interval='@daily',
)


upsert_basic_table_task = BigQueryUpsertTableOperator(
    task_id='upset_basic_table',
    dataset_id='tmdb-movies-410603.dev.movie_basic_info',
    table_resource={
        'tableReference': {'tableId': 'tmdb-movies-410603.dev.movie_details'},
        'expirationTime': (int(time.time()) + 300) * 1000,
    },
)

upsert_numeric_table_task = BigQueryUpsertTableOperator(
    task_id='upset_numeric_table',
    dataset_id='tmdb-movies-410603.dev.movie_numeric_info',
    table_resource={
        'tableReference': {'tableId': 'tmdb-movies-410603.dev.movie_details'},
        'expirationTime': (int(time.time()) + 300) * 1000,
    },
)   

update_basic_table_task
upsert_numeric_table_task

 

 

문제점 : 

• BigQueryUpsertTableOperator 사용 미흡
• GCP 이용을 위해 Airflow 내 Connections 설정하려는데 google_cloud 가 존재하지 않는다.
  • pip install 'apache-airflow-providers-google' 해도 소용없음

 

다음에 할 일 : 

• Airflow 다시 설정해보기

 

참고 : 

• Airflow에 빅쿼리 연동하기 : https://shg-engineer.tistory.com/5
• 빅쿼리 오퍼레이터 사용 예제 : https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_modules/tests/system/providers/google/cloud/bigquery/example_bigquery_dataset.html
• Docker 기반 환경변수로 conntections 관리하기 : https://wooiljeong.github.io/airflow/airflow-manage-env/
• Docker 기반 airflow와 빅쿼리 연동하기 : https://velog.io/@tfj0531/Airflow-와-BigQuery-연결하기-BigQuery-DAG-Example
• Google Cloud Composer로 DAG 만들기 : https://cloud.google.com/composer/docs/data-analytics-googlecloud?hl=ko#console

 

from airflow import DAG
from airflow.providers.google.transfers.bigquery import BigQueryUpsertTableOperator
from datetime import datetime, timedelta
import time

DATASET_NAME = "tmdb-movies-410603.dev"

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 11),
    'end_date': datetime(2024, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('update_analysis_tables', 
        default_args=default_args, 
        description='DAG to update dev. basic, numeric tables',
        schedule_interval='@daily') as dag:

    upsert_numeric_table = BigQueryUpsertTableOperator(
        task_id="upsert_numeric_table",
        dataset_id=DATASET_NAME,
        table_resource={
            "tableReference": {
                "tableId": "movie_numeric_info"
            },
            
        },
        write_disposition="WRITE_TRUNCATE",
        gcp_conn_id='google_cloud_default',
        location='us-west1',  
        provide_context=True,
    )

from airflow import DAG
from airflow.providers.google.transfers.bigquery import BigQueryExecuteQueryOperator
from datetime import datetime, timedelta

DATASET_NAME = "tmdb-movies-410603.dev"

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 11),
    'end_date': datetime(2024, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('update_analysis_tables', 
        default_args=default_args, 
        description='DAG to update dev. basic, numeric tables',
        schedule_interval='@daily') as dag:

    upsert_numeric_table = BigQueryExecuteQueryOperator(
        task_id="upsert_numeric_table",
        sql='''
            MERGE INTO dev.movie_numeric_info AS dest
            USING dev.movie_details AS src
            ON dest.id = src.id
            WHEN MATCHED THEN
              UPDATE SET
                dest.title = src.title,
                dest.popularity = src.popularity,
                dest.budget = src.budget,
                dest.revenue = src.revenue,
                dest.release_date = src.release_date,
                dest.vote_average = src.vote_average,
                dest.vote_count = src.vote_count
            WHEN NOT MATCHED THEN
              INSERT (id, title, popularity, budget, revenue, release_date, vote_average, vote_count)
              VALUES (src.id, src.title, src.popularity, src.budget, src.revenue, src.release_date, src.vote_average, src.vote_count);
        ''',
        use_legacy_sql=False,  # Use standard SQL syntax
        gcp_conn_id='google_cloud_default',
        location='us-west1',
        task_id='upsert_numeric_table',
    )

 

빅쿼리는 FK를 설정할 수 없다. 내일 Airflow 연결 다시 시도해보고 DAG 다시 작성해봐야겠다.

 

최종적으로 사용한 ELT DAG는 다음과 같다.

• 빅쿼리 오퍼레이터 라이브러리를 수정했다.
• Task_id가 두 번 작성되어서 수정했다.

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator
from datetime import datetime, timedelta

DATASET_NAME = "tmdb-movies-410603.dev"

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 11),
    'end_date': datetime(2024, 6, 30),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('update_analysis_tables_v2', 
        default_args=default_args, 
        description='DAG to update dev. basic, numeric tables',
        schedule_interval='@daily') as dag:

    upsert_numeric_table = BigQueryExecuteQueryOperator(
        task_id="upsert_numeric_table",
        sql='''
            MERGE INTO dev.movie_numeric_info AS dest
            USING dev.movie_details AS src
            ON dest.id = src.id
            WHEN MATCHED THEN
              UPDATE SET
                dest.title = src.title,
                dest.popularity = src.popularity,
                dest.budget = src.budget,
                dest.revenue = src.revenue,
                dest.release_date = src.release_date,
                dest.vote_average = src.vote_average,
                dest.vote_count = src.vote_count
            WHEN NOT MATCHED THEN
              INSERT (id, title, popularity, budget, revenue, release_date, vote_average, vote_count)
              VALUES (src.id, src.title, src.popularity, src.budget, src.revenue, src.release_date, src.vote_average, src.vote_count);
        ''',
        use_legacy_sql=False,  # Use standard SQL syntax
        gcp_conn_id='bigquery_dev',
        location='us-west1'
    )

 

 

---

 

6일차

대시보드 연결 및 최종 작업

 

일단 상황 정리를 하자면,

• 문서 및 내용 공유용 노션을 만들어서 공유했다.
• ELT DAG 공유하여 다른분이 Airflow에서 작동되는 것을 확인했다.
  • BigQueryUpsertTableQperator 대신 BigQueryExecuteQueryOperator 작성했다.
  • 이번에는 일단 Airflow 포기.. 다음에 다시 시도해야겠다.
  • 라이브러리 임포트할 때 잘못 작성된 부분, task_id를 두 번 작성한 부분을 수정하니 잘 작동했다고 한다.
  • 다른 데이터 sql문 추가하면 완료!
• ETL DAG는 다른 분이 cloud function 호출 (GCS 증분 생성) -> external 테이블 생성 -> 증분 merge까지 테스트용 DAG 작성 완료
  • Airflow에서 잘 작동되는지를 확인해야 한다.
• 대시보드를 최종 결정해야 한다.

 

1. 대시보드 연결

개인 계정으로 대시보드를 만들어서 팀 공용 preset에 export/import 해야 한다.

1) 대시보드 export : 목록에서 마우스 오버하면 export 아이콘이 뜬다.

2) 우측에 settings 아래에 import 아이콘으로 .zip 파일을 올리면 된다.

 

2. 대시보드 수정

• 상단 : 영화 전체 평균 평점, 분기별 영화 제작 수, 연간 영화 제작 수
• 탭 : 
  • 흥행 분석
  • 장르 인사이트
  • 런타임 분석

대시보드 수정 과정 중에 ETL 할 때 모든 데이터를 가져온다는 것을 확인하였고, 옛날 영화는 부가 정보가 등록되지 않아 분석이 힘들기 때문에 임시로 차트에 필터링을 주었다. 그 외에 ETL이 수정되기도 하여야 한다. 혹시 모를 사태를 대비하기 위해서 차트에도 장치를 해둔 것이다.

 

최종적으로 대시보드 만들어서 공유했고, 이제 다른 사람들 작업 기다리고 문서 작업 등을 하면 된다.

 

 

 

---

 

 

 

프로젝트 결과

 

프로젝트 개요

TMDB를 활용한 영화 흥행 요인 및 트렌드 분석

본 프로젝트는 TMDB 영화 데이터를 활용하여, 데이터 처리와 분석의 전체 과정을 구현하는 것을 목표로 했습니다. 사용된 기술들로는 Google Cloud Platform의 여러 서비스, Apache Airflow, Docker 등이 있으며 이를 통해 실시간 데이터 파이프라인과 시각화 대시보드를 구축했습니다. 핵심 작업은 TMDP API를 통해 데이터를 수집하고, 이를 GCP의 Cloud Storage와 Big Query에 저장한 다음, Airflow를 사용하여 데이터 업데이트를 자동화 하는 것입니다. 데이터는 흥행 요인, 장르별 인사이트, 런타임 분석의 3가지 관점으로 분석되었고, 이를 Preset.io 기반의 대시보드로 시각화하여 사용자가 쉽게 인사이트를 얻을 수 있도록 했습니다.

 

 

프로젝트 결과물

1) 깃 허브

https://github.com/orgs/de-devcourse-prjoject-tmdb/repositories

 

2) 프로젝트 진행 관련 보고서

(추후 업로드)

 

3) 프로젝트 결과물 소개 PPT

(추후 업로드)

 

4) 프로젝트 시연 영상

(추후 업로드)

 

 

피드백

• DAG는 혼자서 파악하고 하기에 쉽지 않다. 팀원들이 이것을 왜 사용하는지를 며칠간 파악해야만 작성할 수 있을 것이다. 따라서 기간이 짧다면 과감하게 포기하고 가는 것도 좋다.

 

---

 

 

회고

탈주자, 잠수자가 발생하여 다소 난항을 겪은 프로젝트여서 아쉽지만, 끝을 냈다는 점에 의의가 있는 프로젝트이다. 흔한 주제를 정했지만 그만큼 새로 배운 기술들을 활용할 수 있는 기회가 더 많아졌다고 생각하고 시작했다. 새롭게 배운 것을 활용하기에 시간을 대부분 쓰고, 이에 대한 고민도 많이 하여 작성한 글들도 많다. 우여곡절을 겪으면서 해결하니 배워가는 것은 많은 것 같다.

 

앞으로 좀 더 협업과 관련된 소프트 스킬에 대해서 생각해보고 싶다. 무엇이 중요하고, 어떤 스타일로 하면 좋은 지 등에 대해서. 그리고 아직 생각하지 못한 부분까지 어우러서 '협업'에 대해 생각하고 정리할 수 있게 되면 좋겠다. 기술적인 부분은 좀 더 공부하면서 채워지겠지만 이렇게 다양한 사람과 협업하는 것은 쉽지 않은 기회일테니 말이다. 그동안 3번의 프로젝트를 하면서 오프라인으로 진행하지 못했고, 온라인으로만 짧게 진행하면서 깨닫고 느낀 점들도 추후 정리해보자!

 

 

---

 

프로젝트 총 할 일

• DW 구축하기 : 빅쿼리
• ETL : TMDB API를 활용하여 수집
• ELT : SQL 구문으로 분석용 테이블 여러 개 생성
• Airflow DAG : ETL, ELT DAG 각각 하나씩 작성
• BI툴에 연동하여 차트/대시보드 생성 : preset.io

 

맡은 일

• ELT : 분석용 테이블 일부분 생성
• Airflow DAG - ELT : ELT DAG 전반적인 부분을 작성
• BI툴에 연동하여 차트/대시보드 생성 (일부 차트 제외)

 

좋았던 점, 잘한 점

• 처음 작성해보는 ELT DAG였지만 잘 작성하였다.
• 데이터베이스와 대시보드툴의 연결을 잘 수행했다. 다운로드되는 파일 형식을 보면서 공부가 되었다.
• 온라인이다보니 겪게 되는 팀플의 불편함이 있는데, 이번에는 잠수(책임 전가)의 문제가 있었다. 이 때 먼저 단체 채널에 현재 상황, 무엇을 하고 있는지 공유하고 질문을 던졌다. 그랬더니 다른 팀원들의 반응이 왔다. 끝내지 못할 수도 있었다고 생각하는데, 소통을 놓치지 않았기 때문에 가능했던 것 같다.

 

아쉬운 점, 부족한 점

• DAG를 챗GPT를 통하여 작성하는 중, 사라진 오퍼레이터를 알려주는 줄도 모르고 난항을 겪였다. >> 공식 문서를 찾아봐야 한다.
• 공식 문서에서 실제 오퍼레이터를 분석하는 능력이 부족하다. 예시가 마땅히 없으면 사용하기 어렵다. 영어 능력도 부족함을 느꼈다.
• 중간에 무단이탈, 잠수타는 팀원들이 있어서 그런지 프로젝트에서 수행한 범위가 대폭 줄어들었다.
• 장르가 여러 개가 합쳐서 하나의 셀에 적혀있는데 이를 전처리하여 활용하는 데에 시간이 많이 걸렸다.
• BI툴이 어떻게 기능을 제공하는지 알아보지 않고 ELT 먼저 했더니, 불필요한 부분도 ELT한 경우가 생겨버렸다.
• GCP에서 서버를 열어 Airflow를 돌렸는데, 그러다보니까 기본적인 부분에서 에러가 발생했다. 아직도 해결을 못 해서 아쉽다. (빅쿼리를 airflow에서 사용하려면 google 관련 세팅을 해야 하는데, 아무리 해도 이 세팅이 추가되지 않는다.)

 

배운 점 (하드스킬, 소프트스킬)

• 어떤 프로젝트건 리드하는 사람이 있어야 할 것 같다. 그렇지 않으면 '누군가 하겠지'라는 마인드가 되기 쉽상이다.
• 사용하고자 하는 툴/기술에 대해서 기초적으로 파악을 해야 한다. 그래야 중복 작업이나 불필요한 작업을 걸러낼 수 있다.
• 공유하는 것은 중요하다. 먼저 말을 꺼내면 누군가 반응한다. 처음이 어렵지, 그 이후 대답은 쉬운 법이다.

 

바라는 점 (보완, 개선하고 싶은 점)

• 프로젝트의 수준이 아쉽다.
• 추후에는 DW 구축, 서버 연결도 도전하고 싶다.
• Airflow 정상 작동하도록 수정해야 한다.