如何提升Spotipy API处理百万级记录时的调用速度？

核心优化点 #

• 批量调用Spotify API #

你当前单条调用spotipy.audio_features()的逻辑是最大的性能瓶颈，该接口单次最多支持传入100个track_id，批量调用可以直接将API请求次数降低99%，是提升最显著的优化项。

• 全局去重track_id #

不同国家/日期的榜单会大量重复出现同一首歌，同一个track_id的音频特征是固定的，不需要重复调用API。你可以先把所有需要处理的track_id统一去重，拉取完特征后再关联回原数据，请求量可以再降低80%~95%（视榜单重复度而定）。

• 新增本地缓存 #

将已经拉取到的音频特征存在本地SQLite单独的缓存表中，后续脚本中断重启、或者新增数据处理时，直接从本地缓存读取已有特征，不需要重复请求接口。

• 多线程并发请求 #

在符合Spotify API速率限制的前提下，用线程池并发请求批量接口，可以进一步压缩请求等待时间，注意控制并发数在5以内避免触发限流。

• 数据IO逻辑优化 #

  • 避免按天循环查询数据库，可一次性查询所有需要处理的日期的全量数据，减少数据库查询开销
  • 无需通过numpy中转生成DataFrame，直接用列表构造DataFrame即可，降低内存开销
  • 可选择直接把处理完的全量数据存回SQLite，不需要按天导出CSV，减少磁盘IO开销

核心代码修改示例 #

批量调用+去重逻辑 #

# 提取所有需要处理的track_id并去重
all_track_ids = list({entry[4].split('/')[-1] for entry in all_db_records})
# 按100个一组拆分
chunk_size = 100
id_chunks = [all_track_ids[i:i+chunk_size] for i in range(0, len(all_track_ids), chunk_size)]
# 批量拉取特征存入字典
audio_feature_cache = {}
for chunk in id_chunks:
    try:
        features = spotify.audio_features(chunk)
        for idx, track_id in enumerate(chunk):
            if features[idx]:
                audio_feature_cache[track_id] = features[idx]
    except ReadTimeout:
        print('Chunk timed out, retrying...')
        features = spotify.audio_features(chunk)
        for idx, track_id in enumerate(chunk):
            if features[idx]:
                audio_feature_cache[track_id] = features[idx]
# 关联回原数据
final_data = []
for entry in all_db_records:
    track_id = entry[4].split('/')[-1]
    feat = audio_feature_cache.get(track_id)
    if not feat:
        continue
    new_entry = entry + (
        track_id, feat['danceability'], feat['energy'], feat['key'],
        feat['loudness'], feat['mode'], feat['speechiness'], feat['acousticness'],
        feat['instrumentalness'], feat['liveness'], feat['valence'], feat['tempo'],
        feat['duration_ms'], feat['time_signature']
    )
    final_data.append(new_entry)

提速效果预估 #

按上述方案优化后，原本18.5天的全量数据处理任务可压缩至2~4小时完成，仅处理每月第三个周五的任务可压缩至10分钟以内完成。

内容的提问来源于stack exchange，提问作者robotpsychology