Datalake : s3, parquet - helloMinji/chatbot_spotify GitHub Wiki

• 필요한 패키지 불러오기

import boto3        # aws sdk
import pandas as pd # 특정 jsonpath만 가져오기 위해
import jsonpath     # pip3 install jsonpath --user

1. top-track API 활용

1. 필요한 key 미리 설정

• jsonpath 패키지를 통해 해당 path 안에서 key에 맞는 value를 반환한다.
• parquet 에러를 피하고, 원하는 값만 가져온다.
  • key: 갖고오고 싶은 값
  • value: api에서의 key, 즉 path (빨간색 표시된 것들, 이미지 참고)

external_urls의 value가 딕셔너리, 그 딕셔너리의 key가 spotify여서 이런 형태로 가져온다.

	top_track_keys = {
		"id": "id",
		"name": "name",
		"popularity": "popularity",
		"external_url": "external_urls.spotify"
	}

2. api로 데이터 불러오기

    top_tracks = []

for문을 돌면서 cursor에서 가져온(코드의 1. 부분) id가 붙은 url이 추가된다.

    for (id, ) in cursor.fetchall():
        URL = "https://api.spotify.com/v1/artists/{}/top-tracks".format(id)
        params = {
            'country': 'US'
        }
        r = requests.get(URL, params=params, headers=headers)
        raw = json.loads(r.text) # api로 가져온 데이터 저장

3. 데이터 업데이트

jsonpath가 추가되면서 생긴 for loop

    for i in raw['tracks']:

• raw라는 dictionary의 tracks라는 key에 해당하는 value가 i값으로 들어간다.
  (핑크색 v 표시된 부분 시작부터 끝까지가 value, 즉 i!)
• i의 타입: dictionary

id, name, popularity, external_url 데이터 업데이트

    top_track = {}
    
    for k, v in top_track_keys.items():
        top_track.update({k: jsonpath.jsonpath(i, v)}

• i라는 dictionary에서 v가 key인 value를 가져와라
• jsonpath의 결과! 그래서 k-jsonpath의 결과가 key-value로 top_track에 업데이트된다!

artist_id 는 이미 갖고 있기 때문에 jsonpath 사용 안 함

        top_track.update({'artist_id': id})  
        top_tracks.append(top_track)
        
    track_ids = [i['id'][0] for i in top_tracks]

4. DataFrame -> Parquet ⭐

• s3에는 json보다 parquet으로 저장하는 것이 더 좋기 때문

    top_tracks = pd.DataFrame(top_tracks)
    top_tracks.to_parquet('top-tracks.parquet', engine='pyarrow', compression="snappy")

• 위에서 필요한 부분만 가져오기 때문에 parquet화 에러 없음
• 보통은 json으로 먼저 저장하고, 새로운 데이터가 들어오는 경우 이미 지정된 형태로 parquet화를 하고, 다른 버킷에 넣는다. 완전 에러 없기 위해!
  👉 이 프로세스를 구축하는 것이 데이터 파이프라인 👈

5. s3에 저장 ⭐

    dt = datatime.utcnow().strftime("%Y-%m-%d")

• utcnow: unix time

불러온 데이터로 파티션 생성

    s3 = boto3.resource('s3')
    object = s3.Object('spotify-artists', 'top-tracks/dt={}/top-tracks.parquet'.format(dt))

• argument1: 버킷 이름
• argument2: 파티션 생성.
  날짜를 기준으로 파티션을 만들어서, 가장 최근 것을 가져오거나 원하는 시기의 데이터를 가져올 수 있게 된다.

    data = open('top-tracks.parquet','rb')
    object.put(Body=data)

2. Audio features API 활용 (batch)

1. 배치 만들기

        tracks_batch = [track_ids[i: i+100] for i in range(0, len(track_ids), 100)] # audio features는 100개까지만 가능

        audio_features = []

2. batch 활용하여 데이터 불러오기

        for i in tracks_batch:
            ids = ','.join(i)
            URL = "https://api.spotify.com/v1/audio-features/?ids={}".format(ids)
            r = requests.get(URL, headers=headers)
            raw = json.loads(r.text)

3. 데이터 업데이트

            audio_features.extend(raw['audio_features'])

4. DataFrame -> Parquet ⭐

        audio_features = pd.DataFrame(audio_features)
        audio_features.to_parquet('audio-features.parquet', engine='pyarrow', compression='snappy')

5. s3에 저장 ⭐

        s3 = boto3.resource('s3')
        object = s3.Object('spotify-artists', 'audio-features/dt={}/audio-features.parquet'.format(dt))
        data = open('audio-features.parquet','rb')
        object.put(Body=data)