在Kaggle中批量下载ZIP链接并合并为CSV的优化方案咨询

需求背景 #

在Kaggle环境中，通过一组年度数据的ZIP文件URL，将数据下载到Python中并合并为单个CSV文件，最终保存到Kaggle工作区或本地机器。现有代码已实现核心功能，以下提供几种更优的实现思路。

现有基础代码 #

url_list = ['https://mapfiles.nrel.gov/data/solar/ae014839fbbe9de5c30bedf56a2f5521.zip', 'https://mapfiles.nrel.gov/data/solar/ea8f39523778ba0223a28116a3e9d85a.zip']

import requests, zipfile, io
import pandas as pd  # 补充原代码遗漏的pandas导入

data_list = []
for url in url_list: 
    r = requests.get(url)
    z = zipfile.ZipFile(io.BytesIO(r.content))
    data_list.append(pd.read_csv(z.open(z.namelist()[0])))

# 合并为总数据表
df = pd.concat(data_list)
df.to_csv('WeatherData.csv')

优化方案及实现 #

1. 并行下载提升处理效率 #

当URL数量较多时，串行下载会耗费大量时间，使用线程池实现并行下载可大幅缩短耗时：

import requests
import zipfile
import io
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

url_list = ['https://mapfiles.nrel.gov/data/solar/ae014839fbbe9de5c30bedf56a2f5521.zip', 'https://mapfiles.nrel.gov/data/solar/ea8f39523778ba0223a28116a3e9d85a.zip']

def load_zip_data(url):
    r = requests.get(url)
    z = zipfile.ZipFile(io.BytesIO(r.content))
    return pd.read_csv(z.open(z.namelist()[0]))

# 线程数可根据URL数量和服务器限制调整
with ThreadPoolExecutor(max_workers=4) as executor:
    data_list = list(executor.map(load_zip_data, url_list))

df = pd.concat(data_list)
df.to_csv('WeatherData.csv', index=False)  # 不保存索引列，减少文件体积

2. 增加异常处理增强健壮性 #

网络请求、文件读取过程中可能出现各类错误，添加异常处理可避免程序崩溃并定位问题：

import requests
import zipfile
import io
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

url_list = ['https://mapfiles.nrel.gov/data/solar/ae014839fbbe9de5c30bedf56a2f5521.zip', 'https://mapfiles.nrel.gov/data/solar/ea8f39523778ba0223a28116a3e9d85a.zip']

def load_zip_data(url):
    try:
        r = requests.get(url, timeout=15)
        r.raise_for_status()  # 校验HTTP请求是否成功
        z = zipfile.ZipFile(io.BytesIO(r.content))
        # 筛选ZIP中的CSV文件，避免非CSV文件导致错误
        csv_files = [file for file in z.namelist() if file.endswith('.csv')]
        if not csv_files:
            raise ValueError(f"URL {url} 的ZIP文件中未找到CSV")
        return pd.read_csv(z.open(csv_files[0]))
    except Exception as e:
        print(f"处理URL {url} 失败: {str(e)}")
        return None

with ThreadPoolExecutor(max_workers=4) as executor:
    data_list = list(executor.map(load_zip_data, url_list))

# 过滤处理失败的数据表
valid_dfs = [df for df in data_list if df is not None]
if valid_dfs:
    df = pd.concat(valid_dfs)
    df.to_csv('WeatherData.csv', index=False)
else:
    print("无有效数据可合并")

3. 流式处理节省内存 #

若单份CSV数据体积较大，一次性加载会占用过多内存，可采用流式追加的方式直接写入最终CSV，无需将所有数据存入内存：

import requests
import zipfile
import io
import pandas as pd

url_list = ['https://mapfiles.nrel.gov/data/solar/ae014839fbbe9de5c30bedf56a2f5521.zip', 'https://mapfiles.nrel.gov/data/solar/ea8f39523778ba0223a28116a3e9d85a.zip']

is_first_file = True
output_path = 'WeatherData.csv'

for url in url_list:
    try:
        r = requests.get(url, timeout=15)
        r.raise_for_status()
        z = zipfile.ZipFile(io.BytesIO(r.content))
        csv_files = [file for file in z.namelist() if file.endswith('.csv')]
        if not csv_files:
            print(f"URL {url} 的ZIP文件无CSV，跳过")
            continue
        # 读取当前CSV并追加到目标文件，仅第一次写入表头
        df = pd.read_csv(z.open(csv_files[0]))
        df.to_csv(output_path, mode='a', header=is_first_file, index=False)
        is_first_file = False
    except Exception as e:
        print(f"处理URL {url} 失败: {str(e)}")

4. Kaggle环境适配 #

在Kaggle中，将文件保存到/kaggle/working/目录可直接在输出面板查看或下载，只需修改保存路径：

# 合并后保存
df.to_csv('/kaggle/working/WeatherData.csv', index=False)

# 流式处理时保存
df.to_csv('/kaggle/working/WeatherData.csv', mode='a', header=is_first_file, index=False)

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