从 Spotify API 调用创建 PySpark 数据帧

如何解决从 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 数据是我想要真正理解的东西。

谢谢！

解决方法

所以你有多个 Python 的列表并且你想要创建一个 Spark 的 DataFrame。您应该使用 SparkSession 的 createDataFrame 方法。

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|
+-----+-----+