从 Spotify API 调用创建 PySpark 数据框 [关闭]答案

【问题标题】：Creating PySpark Dataframe from Spotify API Call [closed]从 Spotify API 调用创建 PySpark 数据框 [关闭]
【发布时间】：2021-02-26 02:22:47
【问题描述】：

我目前正在了解有关 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 数据是我想真正理解的东西。

谢谢！

【问题讨论】：

你遇到了什么问题？
我根本没有遇到错误……但我确实看到了“_corrupted-record”。这就是我不太明白如何解决的问题。

标签： python apache-spark pyspark spotify spotipy

【解决方案1】：

所以你有多个 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|
+-----+-----+

【讨论】：