时间戳

首页 > 解决方案 > 从 Spotify API 调用创建 PySpark 数据框

python - 从 Spotify API 调用创建 PySpark 数据框

问题描述

我目前正在了解有关 pySpark 的更多信息，并希望将一个项目放在一起，将 spotify api 数据放入 spark 数据框中进行分析。

我熟悉如何在 Pandas 中执行此操作，并认为该方法会类似，但我很快意识到这里并非如此。我检查了其他 SO 问题，但仍未找到适合我的方法。

我目前拥有的代码如下所示：

from spotipy.oauth2 import SpotifyClientCredentials, SpotifyOAuth
import datetime
import os

from pyspark.sql import SparkSession, types
from pyspark import SparkContext,SparkConf


# Extract Data
def get_data():

        CLIENT_ID = os.environ.get('SP_CLIENT_ID')
        CLIENT_SEC = os.environ.get('SP_CLIENT_SECRET')

        scope = "user-library-read user-read-recently-played"

        today = datetime.datetime.now()
        yesterday = today - datetime.timedelta(days=1)
        yesterday_unix_timestamp = int(yesterday.timestamp()) * 1000

        client_creds_manager = SpotifyClientCredentials(client_id=CLIENT_ID,client_secret=CLIENT_SEC)
        sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id=CLIENT_ID,client_secret=CLIENT_SEC,scope=scope,redirect_uri='http://localhost:3000/callback'))
        
        return sp.current_user_recently_played(after=yesterday_unix_timestamp, limit=30)

# Transform Data in pyspark
def transform():

    data = get_data()

    song_names = []
    artist_names = []
    played_at = []
    timestamps = []

    for song in data['items']:
        song_names.append(song['track']['name'])
        artist_names.append(song['track']['album']['artists'][0]['name'])
        played_at.append(song['played_at'])
        timestamps.append(song['played_at'][0:10])

    song_dict = {
        'song_names': song_names,
        'artist_names': artist_names,
        'played_at': played_at,
        'timestamps': timestamps
    }

    spark = SparkSession.builder.appName('Spotify').getOrCreate()

    sc = spark.sparkContext

    json_rdd = sc.parallelize(song_dict)

    df = spark.read.json(json_rdd, multiLine=True)

    df.show(truncate=False)

transform()

任何帮助将不胜感激，spark 是一种有趣的学习技术，加载 api 数据是我想真正理解的东西。

谢谢！

标签： pythonapache-sparkpysparkspotifyspotipy

解决方案

所以你有多个 Python 的列表，你想创建一个 Spark 的 DataFrame。您应该使用createDataFrameSparkSession 的方法。

songs = "a b c x y z".split()
names = "i h k l m n".split()
df = spark.createDataFrame(zip(songs,names),['songs','names'])
df.show()

你会得到一个 Spark 的 DataFrame df，如下所示：

+-----+-----+
|songs|names|
+-----+-----+
|    a|    i|
|    b|    h|
|    c|    k|
|    x|    l|
|    y|    m|
|    z|    n|
+-----+-----+

推荐阅读