Python实战：从NASA公开API到Pandas数据清洗完整流程

这两年我把不少时间花在倒腾公开数据上，NASA公开API算是我碰过的数据源里比较“慷慨”的那一类——免费、文档清晰、数据质量高得离谱，毕竟是航天级别的数据源。但“公开”两个字容易让人误会，以为拿回来就能直接用。真正开始用Python读取和处理这些JSON数据之后，你才会发现坑全藏在细节里：字段嵌套、日期格式、分页限制、限流策略、异常重试，哪一步没处理好都只能干瞪眼。

这篇文章是我近半年反复调用NASA API、拿真实数据练手的一份实操复盘，覆盖API Key申请、requests请求、JSON解析、Pandas清洗、限流规避和常见报错排查。适合刚接触Python网络请求的初学者，也适合想找一份相对规范的真实数据源做练习的开发者。读完你至少能独立完成一套“请求→解析→清洗→落盘”的完整流程，后面换成别的公开API也只是换皮不换骨。

1. 整体设计：NASA API到底在解决什么问题

1.1 为什么选Python而不是别的工具

先聊一个很多人忽略的问题：抓取公开API，工具多得是，为什么我始终建议用Python？

最直接的原因是生态。NASA的API返回的是JSON格式数据，而JSON在Python里就是字典和列表的组合，json库解析后可以直接用键值对访问，几乎不需要额外的类型转换成本。配合pandas，一个嵌套很深的JSON结构也能在几步之内变成干净整洁的表格。如果用Java或者C#，光是把JSON映射成强类型对象就要写一堆类定义，对这个体量的数据迁移来说性价比太低了。

另一个原因在requests库。它把URL请求、请求头、超时、会话复用这些底层网络细节全部封装掉了。大学的时候用Python写爬虫，用Java的HttpClient写原生请求，光SSL证书验证和重定向处理就能耗掉半章文档。用requests，一个get方法加几个参数就完事。

当然，如果是纯数据分析场景，GraphQL或REST工具也能做，比如Postman先调试请求、再导出代码，但那更适合一次性操作。做成脚本、定时跑、集成到更大的数据处理管道里，还是Python最顺手。

1.2 NASA公开API里值得用的数据源有哪些

NASA开放的数据接口其实远不止一个，很多人只听说过“每日一天文图”（APOD），其实整个API门户下面挂了十几个模块。我实际用下来，比较值得练手的是下面这几个：

这几个模块有一个共同特点：数据返回延迟低，通常在几十到几百毫秒；JSON结构相对稳定，不会三天两头变字段名。对初学者来说这意味着调试成本低，出错时能很快定位是代码问题还是数据源问题。而像卫星轨道这类数据，还涉及时间序列和物理单位换算，用来练手刚好能锻炼数据处理能力。

1.3 整体流程怎么拆

把整个流程拆开看，其实所有API读取和处理项目都可以划分成五个环节：

• 获取访问凭证：申请API Key，明确访问限制，把Key安全地配置到项目中。
• 构造HTTP请求：确定请求URL、查询参数、超时、请求头。
• 解析响应：检查HTTP状态码，解析JSON，处理嵌套结构。
• 数据清洗：把解析后的数据整理成结构化表格，处理缺失值和异常值。
• 持久化：写入CSV、Excel或数据库，方便后续分析和可视化。

这套流程跟爬虫的逻辑有一点区别：API是别人规定好的接口，你必须按契约来请求；网页爬虫则是先有页面、再自己找解析规则。所以API数据处理更像是“按约定取数”，对字段结构的理解比请求技巧更重要。理解了这一点，后面写代码就不容易跑偏。

2. 核心细节：API Key申请、鉴权与基础请求

2.1 一步步申请API Key及参数说明

在NASA API官网上，新用户可以直接申请一个API Key，流程非常简单。第一步是打开API申请页面，填写姓名和邮箱地址，系统会提示你把申请Key时填写的邮箱地址，需要和请求中的邮箱参数保持一致，否则会被拒绝。然后点击生成按钮，页面会返回一串字母数字组合的字符串，这就是token。

这个Key拿到之后，所有请求都需要在查询参数里带上api_key字段。如果没有Key，可以暂时用demo_key代替，但演示Key并发请求受限、频率限制也更容易触发，而且部分接口可能会拒绝服务。所以正经项目里不要偷懒，花几分钟申请一个免费Key，后面能少很多皮肉之苦。

我处理Key的方式是写进环境变量，而不是硬编码在代码里。这样做的原因很现实：代码大概率会在多个环境跑，本地测试、服务器部署、定时任务，每个环境的Key可能不同；而且如果项目发布到公开仓库，暴露的Key会让别人冒用你的配额。用环境变量加载，代码里永远只留变量名，访问密钥不出现在源码里。

2.2 请求构造的常见坑

请求NASA API时，最容易踩的坑不是API本身复杂，而是对HTTP请求细节不够敏感。

第一个坑是把参数直接拼进URL字符串。比如https://api.nasa.gov/planetary/apod?api_key=xxx&date=2024-01-01。这样做在参数少的时候没问题，一旦参数多了、带特殊字符（比如Sol日期、经纬度、百分号），URL编码很容易出错。用requests.get(url, params={...})的好处是库会自动帮你做URL编码和参数拼接，你只负责维护一个干净的字典。

第二个坑是忽略超时设置。很多初学者直接requests.get(url)不带超时，一旦网络波动，请求会一直挂着，程序看起来像死循环。NASA API偶尔会在节假日或者大型任务期间响应变慢，甚至几分钟不响应，这时候超时设置就是保命动作。我一般设置timeout=(3.05, 10)，前一个是连接超时，后一个是读取超时，根据网络状况微调。

第三个坑是不检查HTTP状态码就强行解JSON。API返回200不代表你要的数据就在里面，返回429、500才是常态。正确做法是先判断response.ok，再用内置JSON解析器，异常时打印出响应体内容，方便定位问题。

2.3 会话复用与限流策略

NASA API的标准限制是每小时1000次请求，单日约25000次。听着挺多，但如果你在循环里逐条请求数据（比如按日期逐日拉取两年数据），很轻松就能把配额塞满。

我在并发请求上都用requests.Session()管理连接。Session能复用TCP连接，重复请求同一域名时会显著减少握手开销，速度提升不大，但能减轻服务器压力，不容易触发封禁。同时利用time.sleep()控制请求间隔，这个方案简单可靠，比复杂令牌桶实现更直观。

一个不太常用的技巧是提前分析响应头里的X-RateLimit-Remaining字段。NASA返回头里会带当前剩余配额，把它解析出来，程序就知道自己还剩多少额度，可以在接近限额时自动放慢速度或停下来。这个信息在日志里打出来，每次跑脚本心里都有底。

3. 实操过程：三套完整示例，从拉数据到出结果

3.1 示例一：抓取APOD每日天文图片元数据

APOD是NASA公开API里最简单的接口，也是我建议新手拿它练手的原因。它返回的数据量小、字段明了，非常适合用来走通整个HTTP请求流程。

先写一个基础版本：

import requests import json API_KEY = "你的key" def fetch_apod(date=None): url = "https://api.nasa.gov/planetary/apod" params = {"api_key": API_KEY} if date: params["date"] = date response = requests.get(url, params=params, timeout=(3.05, 10)) response.raise_for_status() data = response.json() return data if __name__ == "__main__": data = fetch_apod("2024-06-01") print(json.dumps(data, indent=2, ensure_ascii=False))

这段代码核心就两个动作：构造params字典、用raise_for_status()兜底。拿到返回的JSON后，你能看到title、explanation、url、date、media_type这些字段。其中media_type可能是image或video，这个字段很多人不看，导致后面统一保存图片时通用逻辑会报错。

继续扩展，把抓取逻辑改成批量，并落到CSV：

import csv import time from datetime import date, timedelta def batch_fetch_apod(start_date, days=30): results = [] current = start_date for _ in range(days): try: item = fetch_apod(current.isoformat()) results.append({ "date": item.get("date"), "title": item.get("title"), "media_type": item.get("media_type"), "url": item.get("url"), "hdurl": item.get("hdurl"), }) except Exception as exc: print(f"{current} 抓取失败: {exc}") time.sleep(0.5) # 低于限频，避免触发限流 current += timedelta(days=1) return results def save_to_csv(rows, filename="apod_data.csv"): with open(filename, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows)

这里有几个细节值得解释。第一，start_date.isoformat()会把datetime对象输出成YYYY-MM-DD格式，正好符合NASA API的日期参数要求。第二，每次请求之间加0.5秒延时，是为了留出余量，避免短时间内高频请求触发限流。第三，异常处理放在单条记录上，而不是整个循环外，否则一条记录出错会导致整体中断。第四，保存CSV时用newline=""，避免Windows平台在行尾多出空行。

这个示例跑通后，你等于掌握了绝大多数NASA API的基础请求模式。

3.2 示例二：火星车照片按日期批量拉取与Pandas清洗

火星车照片接口比APOD复杂不少，它的核心查询参数是earth_date（地球日期）或sol（火星日），返回的数据是一个嵌套结构：外层是照片列表，每张照片里又包含火星车信息、相机信息和图片URL。

我用这个接口做批量拉取时，最喜欢用Pandas处理，因为数据天然可以拉平成表格：

import pandas as pd import requests import time from datetime import date, timedelta API_KEY = "你的key" BASE_URL = "https://api.nasa.gov/mars-photos/api/v1/rovers/curiosity/photos" def fetch_photos_by_date(earth_date, page=1): params = { "api_key": API_KEY, "earth_date": earth_date, "page": page, } response = requests.get(BASE_URL, params=params, timeout=(3.05, 10)) response.raise_for_status() return response.json().get("photos", []) def collect_photos(start_date, days=7): all_photos = [] current = start_date for _ in range(days): page = 1 while True: photos = fetch_photos_by_date(current.isoformat(), page) if not photos: break all_photos.extend(photos) page += 1 time.sleep(0.4) current += timedelta(days=1) return all_photos raw_photos = collect_photos(date(2024, 6, 1), days=7) df = pd.json_normalize(raw_photos) # 关键一步：嵌套字典自动展平

pd.json_normalize是这里最值得圈出来的函数。它会把嵌套的camera字段展平成camera.id、camera.name、camera.full_name这样的列，把rover对象里的rover.name、rover.status也变成独立列。如果没有这个函数，你得自己写循环拆字段，代码又臭又长还容易漏字段。

展平之后，表格里可能出现不少用不到的列，比如rover.max_sol、rover.total_photos。我的做法是直接选列，只保留有分析价值的字段：

cols = ["id", "sol", "img_src", "earth_date", "camera.full_name", "rover.name"] df_clean = df[cols].copy() df_clean.columns = ["photo_id", "sol", "image_url", "earth_date", "camera_name", "rover_name"]

注意这里我用的是df[cols].copy()，不是df[cols]。原因是后者返回的是视图，后续修改数据时可能触发SettingWithCopyWarning，用copy()就把数据和原表彻底切开了，后续想怎么改都不影响原数据。

火星车照片里那个img_src字段其实还藏着一个坑：照片链接对应的图片文件不一定还在NASA的服务器上，有些早期照片的链接已经失效。所以如果你想批量下载图片，务必在下载循环里捕获HTTP异常，不要把“API返回了照片列表”误当成“照片一定可以访问”。

3.3 示例三：近地天体数据分页拉取与字段展平

NeoWs接口返回的是近地小行星的轨道和接近信息，它的JSON嵌套程度比火星车照片更深，而且自带分页逻辑。这个接口非常适合用来练“多级数据处理”。

请求示例：

API_KEY = "你的key" BASE_URL = "https://api.nasa.gov/neo/rest/v1/neo/browse" def fetch_neo_page(page): params = {"api_key": API_KEY, "page": page, "size": 20} response = requests.get(BASE_URL, params=params, timeout=(3.05, 10)) response.raise_for_status() data = response.json() return data["near_earth_objects"], data["page"] all_neo = [] page = 0 total_pages = 1 while page < total_pages: items, page_info = fetch_neo_page(page) all_neo.extend(items) total_pages = page_info.get("total_pages", 1) page += 1 time.sleep(0.3)

这个接口翻页是从0开始的，跟很多API从1开始不一样，我第一次写的时候就掉进去了，结果第一页数据重复拉取。这个细节在代码注释里一定要写清楚，不然过两个月自己看代码也会懵。

拿到的all_neo是一个列表，每个元素是类似这样的结构：id、name、absolute_magnitude_h、estimated_diameter（嵌套着公里、英里、米、英尺四个单位）、is_potentially_hazardous_asteroid、close_approach_data（一个列表，包含接近日期、相对速度、距离等）。

用户真正关心的往往不是全部字段，而是“这个天体有没有威胁、什么时候最近、距离多近、速度多快”。所以展平要按需处理：

import pandas as pd def flatten_neo(item): close_approach = item["close_approach_data"][0] estimated_km = item["estimated_diameter"]["kilometers"] return { "neo_id": item["id"], "name": item["name"], "abs_magnitude": item["absolute_magnitude_h"], "est_diameter_km_min": estimated_km["estimated_diameter_min"], "est_diameter_km_max": estimated_km["estimated_diameter_max"], "hazardous": item["is_potentially_hazardous_asteroid"], "close_time": close_approach["close_approach_date"], "relative_velocity_km_s": close_approach["relative_velocity"]["kilometers_per_second"], "miss_distance_km": close_approach["miss_distance"]["kilometers"], "orbiting_body": close_approach["orbiting_body"], } df_neo = pd.DataFrame([flatten_neo(item) for item in all_neo])

这段函数里做了一个显式假设：取close_approach_data的第一条记录，也就是最近的一次接近数据。这个假设在大多数场景下成立，但严格来说应该让它有据可循，所以注释里最好写上“这里只取最近一次接近记录，如需全量可以遍历”。数据来源足够规范，但处理逻辑不能想当然。

清洗完之后还能顺手做一个简单统计：有多少颗属于“潜在威胁天体”，平均相对速度是多少。这一步用Pandas一行代码就搞定，真正把“读取”升级到了“处理”。

4. 常见问题与排查技巧实录

4.1 HTTP状态码与限流问题

所有公开API都会遇到状态码问题，NASA API也一样。我把实际踩过的坑整理成了一个速查表：

遇到429是最常见的。我的处理方式不是等死，而是写一个带重试的请求包装器：

import time import requests def get_with_retry(url, params, max_retries=3, base_sleep=2): for attempt in range(max_retries): try: response = requests.get(url, params=params, timeout=(3.05, 10)) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", base_sleep)) time.sleep(retry_after) continue response.raise_for_status() return response except requests.exceptions.RequestException as exc: print(f"第{attempt+1}次请求失败: {exc}") time.sleep(base_sleep * (attempt + 1)) raise RuntimeError("重试多次仍然失败")

这个包装器做了两件重要的事情。第一，显式读取Retry-After响应头，按服务器要求的时间等待，而不是自己拍脑袋等多少秒；第二，用指数退避策略处理非429的临时错误，第一次等2秒、第二次等4秒、第三次等6秒，给服务器留出恢复窗口。

很多初学者遇到429就直接放弃或者无限重试，前者浪费了之后可能恢复的数据，后者会让服务器把你当成恶意请求者，反而封禁更久。正确的思路是“有节奏地重试”，这是公开API数据处理的基本素养。

4.2 JSON解析与字段缺失的坑

NASA API的JSON结构虽然稳定，但不代表每个返回对象都包含所有字段。我在处理火星车照片时发现，有些照片记录缺少img_src，有些近地天体对象的close_approach_data是空列表。如果用item["close_approach_data"][0]直接取值，IndexError马上就会打断整个脚本。

解决办法是使用.get()方法并设置默认值，或者先用if判断再取数：

if item.get("close_approach_data"): first_approach = item["close_approach_data"][0] else: first_approach = {}

另一个坑是字段值类型的不一致。比如is_potentially_hazardous_asteroid在某些记录里是布尔值，在另一些记录里因为解析原因可能变成字符串"true"。处理时统一转一次类型最保险。

还有日期格式的问题。API返回的日期多数是YYYY-MM-DD或ISO 8601完整时间戳格式。如果你要按时间过滤或分组，千万别直接用字符串比大小。正确做法是：

pd.to_datetime(df["close_time"])

这样Pandas会帮你把字符串解析成datetime类型，之后做时间范围筛选、按月聚合这些操作就方便多了。

4.3 数据保存与图片下载的细节

数据清洗完，最后一个环节是持久化。我试过CSV和Excel两种格式，大部分场景CSV够用，但有几个细节不能不管。

编码问题。Windows下用Pandas写CSV，默认编码可能是GBK，直接用Excel打开没问题，但换到Linux或Mac上就会乱码。我统一用utf-8-sig编码，既保证跨平台兼容，又让Excel正常识别中文：

df.to_csv("neo_data.csv", index=False, encoding="utf-8-sig")

大字段问题。NASA有些字段的值特别长，比如explanation文本、URL，写CSV时容易带出换行符，破坏行结构。处理办法是写CSV前手动替换换行符：

df["explanation"] = df["explanation"].str.replace("\n", " ")

如果你要顺便下载火星车照片，我强烈建议只下载本地的缩略图尺寸或者先看文件大小再决定，否则几百张原图一下来，磁盘空间和带宽都会吃紧。一个可行的做法是设置图片大小参数或者干脆只保留元数据，等真正需要用图的时候再按需下载。

5. 一次完整的自动化拉取脚本拼装

四个步骤都熟练之后，我一般会把这些逻辑拼装成一个完整脚本，让它可以一条命令跑完全流程。下面是一个简化版，把APOD拉取、火星车照片、近地天体三个数据源统一起来处理：

import pandas as pd import requests import time from datetime import date, timedelta API_KEY = "你的key" SESSION = requests.Session() def fetch_json(url, params, max_retries=3): for attempt in range(max_retries): try: response = SESSION.get(url, params=params, timeout=(3.05, 10)) if response.status_code == 429: time.sleep(int(response.headers.get("Retry-After", 2))) continue response.raise_for_status() return response.json() except requests.exceptions.RequestException as exc: print(f"请求失败: {exc}") time.sleep(2 * (attempt + 1)) return None def collect_apod(start_date, days=7): rows = [] for i in range(days): target = start_date + timedelta(days=i) data = fetch_json( "https://api.nasa.gov/planetary/apod", {"api_key": API_KEY, "date": target.isoformat()} ) if data: rows.append({ "source": "APOD", "title": data.get("title"), "date": data.get("date"), "url": data.get("url"), }) time.sleep(0.3) return rows def collect_mars(start_date, days=7): rows = [] for i in range(days): target = start_date + timedelta(days=i) data = fetch_json( "https://api.nasa.gov/mars-photos/api/v1/rovers/curiosity/photos", {"api_key": API_KEY, "earth_date": target.isoformat(), "page": 1} ) for photo in (data or {}).get("photos", [])[:5]: rows.append({ "source": "MARS", "title": photo["rover"]["name"], "date": photo.get("earth_date"), "url": photo.get("img_src"), }) time.sleep(0.3) return rows def collect_neo(): data = fetch_json( "https://api.nasa.gov/neo/rest/v1/neo/browse", {"api_key": API_KEY, "page": 0, "size": 20} ) rows = [] for item in (data or {}).get("near_earth_objects", [])[:10]: rows.append({ "source": "NEO", "title": item.get("name"), "date": item.get("close_approach_data", [{}])[0].get("close_approach_date"), "url": item.get("nasa_jpl_url"), }) return rows if __name__ == "__main__": start = date(2024, 6, 1) all_rows = collect_apod(start) + collect_mars(start) + collect_neo() df = pd.DataFrame(all_rows) df.to_csv("nasa_summary.csv", index=False, encoding="utf-8-sig") print(df.head(10))

这个脚本设计的核心思想是统一返回结构。虽然三个API的原始数据结构完全不同，但我在拼装时都映射成了source、title、date、url四个字段，这样下游的存储和分析逻辑完全不用针对每个数据源写分支。表面上看是损失了一些细节字段，但换来的是极高的可维护性。如果你的分析场景需要完整字段，可以保留原始JSON和摘要字段并行存储。

脚本里的SESSION全局复用也是刻意的。Session对象在脚本内所有请求之间共享，连接复用率提高，请求耗时会从每次几百毫秒降到几十毫秒量级。长时间跑批量任务时，这个优化积累的效果非常明显。

最后还要提一个运维上的习惯：写日志。脚本跑完只打印一个df.head()是不够的，真正的运维需要知道每次跑到哪一天失败了、哪个接口失败了、重试了几次。简单的print(f"当前处理: {target}")就能让问题定位时间从半小时缩短到两分钟。别小看这个习惯，数据量大的时候全靠日志救命。

我个人的体会是，NASA公开API最大的价值不在数据本身，而在于它提供了一个难度适中的真实数据集让你练习完整的数据处理链路——请求、解析、清洗、存储、异常处理，每个环节都有真实的坑等着你去踩。踩一遍，比看十遍文档有用。如果你能自己动手把这个流程完整跑通，后面再去对接气象、金融、社交媒体等公开API，基本就是同一套套路，换个参数名而已。