公共自行车数据这件事,我盯了很久。原因很简单:公共自行车站点数据是少有的“接口足够开放、数据量适中、业务含义清晰”的爬虫练手对象——它不像电商平台那样反爬严格,也不像社交媒体那样涉及用户隐私红线,而且站点分布、车辆状态、时间变化这些维度叠加起来,能做的分析花样非常多。这个项目我从采集端到分析端完整跑了一遍,中间踩了不少坑,也沉淀出一套可以复用的方法。这篇内容就是把整个项目从零到一拆开讲清楚,适合那些已经掌握Python基础语法、想找真实项目练手爬虫和数据分析的读者。
1. 项目整体设计与技术选型思路
1.1 为什么选公共自行车数据做爬虫实战
选练手项目有个原则:数据太简单没收获,数据太复杂容易劝退。公共自行车数据刚好落在甜点上。它的核心数据是站点信息——站点编号、名称、经纬度、当前车辆数、可用桩位数,这些字段既覆盖了典型的JSON结构解析场景,又不会像多级评论那样嵌套到怀疑人生。
另一个原因是数据的时间属性。公共自行车数据是典型的“时空数据”,同一个站点在早高峰和深夜的状态截然不同,这给数据分析留足了发挥空间。你可以做站点热度排行,可以做潮汐现象分析,甚至可以结合天气数据看雨天对骑行量的影响——这些都是在“数据采集”之上自然延伸出来的价值点。
还有一点很实际:公共自行车平台大多提供了面向用户的查询接口,这些接口在合规前提下用于个人学习研究是行业常见做法。这意味着爬虫的“头”不会被立马封死,新手能有一个相对友好的环境去理解请求、解析、存储、分析这条完整链路。
1.2 技术栈选型与整体架构
我的技术选型以轻量、易复现为原则:
| 环节 | 选型 | 理由 |
|---|---|---|
| 数据采集 | Python + requests | 简单直接,session管理方便 |
| 解析处理 | json + pandas | JSON天然匹配接口数据,pandas处理表格化数据高效 |
| 数据存储 | SQLite + CSV | 零配置,单文件,适合个人项目 |
| 可视化 | matplotlib + pyecharts | 兼顾静态图表和交互式地图展示 |
| 定时调度 | 本地cron / 手动脚本 | 个人项目不需要上Celery之类重型工具 |
架构上就是三条线:采集线负责定时拉取站点数据并落库,清洗线负责把原始JSON转成规整的表结构并处理异常值,分析线负责从时间、空间、站点三个维度做探索性分析和可视化。三条线解耦,哪一环出了问题可以单独排查。
这里有个教训:一上来不要追求分布式爬虫或者Scrapy框架。requests + 单线程在这个场景下完全够用,公共自行车站点数据量级也就是几百个站点、每次请求几十KB,把基础链路跑通比追求架构炫技重要得多。
2. 数据采集目标分析与接口定位
2.1 明确要采集什么字段
动手写代码之前,先想清楚“我到底要采什么”。这决定了后续的存储结构和分析上限。我最终定的字段清单如下:
- 站点编号(唯一标识,做关联主键)
- 站点名称(业务理解用)
- 经度、纬度(空间分析基础)
- 当前可用车辆数(核心业务指标)
- 当前可用空桩数(还车能力指标)
- 采集时间戳(时间分析维度)
- 站点总容量(计算饱和度用)
字段之间不是孤立的。可用车辆数和可用空桩数加起来通常等于总容量,这个关系可以用来做数据质量校验——如果发现两者之和与容量对不上,说明数据有问题,要么是接口字段含义变了,要么是采集链路中出现了脏数据。
2.2 定位公共自行车数据接口
公共自行车平台的接口定位有两条路。第一条路是直接找官网或App端的数据接口,通过浏览器开发者工具或抓包工具查看网络请求;第二条路是留意一些城市的数据开放平台,部分城市会把公共自行车站点数据作为开放数据公开。
我实际采用的是第一种方式。以国内某公共自行车平台的App接口为例,其站点列表接口通常返回类似这样的JSON结构:
{ "code": 200, "data": { "stations": [ { "station_id": "SZ001", "station_name": "市民中心东门", "longitude": 114.057868, "latitude": 22.543099, "bikes_available": 12, "slots_available": 28, "capacity": 40, "update_time": "2024-11-10 08:30:00" } ] } }当然,不同城市的字段命名差异很大,有的叫available_bikes,有的叫bike_count,还有的干脆是拼音缩写。这些都需要在实际抓取时先打印出原始响应,对照字段含义逐个确认。我建议把“打印原始返回”这一步做成习惯——很多人一上来就套解析逻辑,字段对不上时排查起来非常痛苦。
提示:抓包分析接口时,只看数据内容和个人学习用途,不要尝试绕过任何权限控制、批量抓取用户信息或高频请求影响平台稳定性。合法合规是底线。
2.3 接口请求策略与反爬应对
公共自行车数据接口的反爬强度通常不高,但并不意味着可以肆无忌惮地请求。我的实践是用一个“温和”的请求策略:
import requests import time session = requests.Session() headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "application/json", "Referer": "https://example-bike-platform.com/" } def fetch_station_data(url, interval=2): try: resp = session.get(url, headers=headers, timeout=10) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None finally: time.sleep(interval)interval=2是我反复试验后的平衡点。公共自行车数据每分钟甚至每30秒就会变化,但个人分析完全不需要那么高的采集频率,5分钟一次已经能看出明显的时间趋势了。控制请求频率既是对目标服务器的尊重,也是降低自己IP被临时限制的风险。
实际踩坑中发现,有些平台会校验Referer字段,如果不带上正确的来源页面引用,接口会返回403。这种情况只要在headers里补上从浏览器复制来的Referer就能解决。另外,少数平台会做请求频率限制,连续高频请求后返回“请求过于频繁”的提示,解决办法就是降频加重试。
3. 爬虫核心实现与数据落库
3.1 从requests到数据解析的完整链路
单次请求拿到的是JSON文本,要变成可分析的结构化数据,中间需要经过一层“映射清洗”。我的做法是每次请求后立即做一次字段标准化,把不同平台的差异字段统一成自己的命名规范:
import pandas as pd from datetime import datetime def parse_station_data(json_data, source="city_bike"): stations = json_data.get("data", {}).get("stations", []) records = [] fetch_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for item in stations: records.append({ "station_id": item.get("station_id") or item.get("id"), "station_name": item.get("station_name") or item.get("name"), "longitude": item.get("longitude") or item.get("lng"), "latitude": item.get("latitude") or item.get("lat"), "bikes_available": item.get("bikes_available") or item.get("bike_count"), "slots_available": item.get("slots_available") or item.get("empty_count"), "capacity": item.get("capacity") or item.get("total"), "fetch_time": fetch_time, "source": source }) return records注意这里用了or来做字段兼容,但有个隐患:如果某个字段的值为0,or会把0也替换掉。比如bikes_available为0时,item.get("bikes_available") or item.get("bike_count")如果前者不存在才走后者,如果前者存在但值为0,会直接返回0,不会走到or后面的逻辑——这里其实没问题。但如果平台某个字段缺失返回None,or会尝试下一个备选字段。不过更严谨的做法是显式判断is not None,避免字段值为空字符串或0时出现误判。这个细节在处理真实数据时非常关键。
3.2 数据存储设计:SQLite与CSV双轨
存储方案上我采用了SQLite为主、CSV为辅的双轨制。SQLite负责存储历史累积数据,便于后续按时间范围查询和分析;CSV作为即时快照,方便用Excel快速查看某次采集的结果。
建表语句如下:
CREATE TABLE IF NOT EXISTS station_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, station_id TEXT NOT NULL, station_name TEXT, longitude REAL, latitude REAL, bikes_available INTEGER, slots_available INTEGER, capacity INTEGER, fetch_time TEXT NOT NULL, source TEXT ); CREATE INDEX IF NOT EXISTS idx_station_time ON station_status (station_id, fetch_time);索引idx_station_time很重要。如果数据累积到几万条甚至几十万条,不带索引做时间范围和站点的联合查询会明显变慢。刚开始我觉得数据量小不需要索引,后来做一个月的趋势分析时查询时间从秒级涨到了十几秒,加上这个复合索引后瞬间回到毫秒级。
写入逻辑用executemany批量插入,比逐条execute快一个数量级:
import sqlite3 conn = sqlite3.connect("bike_data.db") cursor = conn.cursor() def save_to_sqlite(records): sql = """ INSERT INTO station_status (station_id, station_name, longitude, latitude, bikes_available, slots_available, capacity, fetch_time, source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """ cursor.executemany(sql, [ (r["station_id"], r["station_name"], r["longitude"], r["latitude"], r["bikes_available"], r["slots_available"], r["capacity"], r["fetch_time"], r["source"]) for r in records ]) conn.commit()3.3 异常重试与日志记录机制
爬虫跑起来后最大的敌人不是反爬,而是网络抖动和接口异常。我遇到过好几次半夜定时任务跑挂的情况:某次请求超时没有捕获,整个脚本直接崩溃退出,第二天才发现数据缺了好几个小时。
解决思路有两条:
第一,请求层面加重试机制,对502、503、超时这类临时性错误做指数退避重试:
def fetch_with_retry(url, headers, max_retries=3): for attempt in range(max_retries): try: resp = requests.get(url, headers=headers, timeout=10) if resp.status_code == 200: return resp.json() elif resp.status_code in (502, 503, 504): time.sleep(2 ** attempt) continue else: return None except requests.exceptions.Timeout: time.sleep(2 ** attempt) return None第二,脚本入口处加日志记录,每条任务开始和结束都写入日志文件。这样即使出了问题,也能快速定位是请求层、解析层还是存储层出的故障。我的日志格式非常简单:[时间] [级别] 消息,手动实现不到二十行,但排查问题时价值巨大。
4. 数据分析实战:从清洗到可视化
4.1 数据清洗与特征加工
原始数据落库后不能直接分析,原因有三个:字段值可能异常(车辆数大于容量)、时间字段是字符串需要转datetime、经纬度可能存在缺失值。清洗流程我用pandas完成:
import pandas as pd df = pd.read_sql_query("SELECT * FROM station_status", conn) df["fetch_time"] = pd.to_datetime(df["fetch_time"]) df["hour"] = df["fetch_time"].dt.hour df["weekday"] = df["fetch_time"].dt.weekday # 异常值过滤:可用车辆数不能超过容量 df = df[df["bikes_available"] <= df["capacity"]] # 容量为空时填充为0 df["capacity"] = df["capacity"].fillna(0) # 新增饱和度特征 df["saturation"] = df["bikes_available"] / df["capacity"].replace(0, np.nan)这里新增的两个特征——hour和weekday——是后续所有时间维度分析的基础。saturation(饱和度)表示站点当前的“满车程度”,值越高说明可用车越多,值越低说明车基本被借光了。
这里有一个比较隐蔽的问题:公共自行车站点在深夜时段容量可能被调整为0(部分站点夜间关闭),这种情况下bikes_available / capacity会出现除零。我用replace(0, np.nan)把除零结果变成缺失值,后续分析时统一忽略这些记录。
4.2 时空维度分析:站点热度与潮汐规律
清洗完成后,第一个值得做的分析是“全城站点车辆数的时间曲线”。这个曲线可以直接反映公共自行车系统的整体运行节奏:
hourly_stats = df.groupby("hour")["bikes_available"].mean().reset_index()从结果可以看到典型的双峰曲线:早上7点到9点,整体可用车辆数下降(大家骑车去地铁站);晚上18点到20点,可用车辆数回升(大家骑回家)。这个现象背后的业务逻辑是通勤潮汐——公共自行车主要承担“最后一公里”接驳功能。
更进一步,可以对单个站点做“24小时饱和度热力图”。把站点作为行、小时作为列、饱和度作为值,用pivot_table就能得到,然后配合seaborn画热力图:
pivot = df.pivot_table( index="station_name", columns="hour", values="saturation", aggfunc="mean" )热力图能非常直观地暴露“潮汐站点”。我实际的例子中,某个地铁站旁边的站点早高峰饱和度只有0.1(车全被骑走了),晚高峰饱和度达到0.9(车全停回来了)。这种站点就是调度公司最需要关注的“失衡点”。
4.3 空间可视化:用pyecharts画站点分布图
公共自行车数据的空间属性如果不做可视化,会浪费一半价值。我推荐用pyecharts的散点图来做站点分布地图:
from pyecharts import options as opts from pyecharts.charts import Scatter scatter = ( Scatter() .add_xaxis(df["longitude"].tolist()) .add_yaxis("站点", df["latitude"].tolist(), label_opts=opts.LabelOpts(is_show=False)) .set_global_opts( visualmap_opts=opts.VisualMapOpts( dimension=0, min_=df["saturation"].min(), max_=df["saturation"].max() ) ) ) scatter.render("station_map.html")站点分布图能快速看出站点的空间覆盖密度——哪些区域站点密集、哪些区域是盲区。如果再叠加“平均饱和度”作为颜色维度,就能直观地看到哪些片区的车辆长期过剩或长期不足。
这块我还尝试过用folium生成带底图的交互式地图,效果更好但需要互联网底图服务。pyecharts的好处是生成的HTML可以离线打开,且视觉风格统一,作为项目交付物很方便。
5. 常见问题与排查技巧实录
5.1 接口返回乱码或解析失败怎么办
最常见的坑是编码问题。有些平台的接口返回的不是标准UTF-8,而是GBK或GB2312编码。requests库默认按响应头猜测编码,但如果接口没有正确声明charset,就会拿到乱码。
解决办法是在拿到响应后手动指定编码:
resp = requests.get(url, headers=headers) resp.encoding = "utf-8" # 或改为 resp.encoding = "gbk" data = resp.json()判断该用哪种编码,先看resp.apparent_encoding,这个字段是requests根据字节内容推测的编码,通常比较可靠。如果还不行,打印前500字节的原始内容,看到\\xe4\\xb8\\xad这种格式基本是UTF-8,看到\\xd6\\xd0这种格式大概率是GBK。
JSON解析失败的另一个可能原因是接口返回了非JSON内容,比如一个HTML错误页或者一段JavaScript。这种情况要先打印resp.text[:500]看看返回的到底是什么,再对症下药。
5.2 定时任务跑挂,数据出现缺口
我在实践中最常遇到的问题就是定时任务静默失败。cron任务执行时如果脚本抛出未捕获异常,cron不会发邮件通知,只会静默退出,第二天早上看数据库才发现漏采了。
对策有三个:
- 脚本入口用
try...except包住整体逻辑,异常信息写入日志文件 - 日志文件按天切割,这样能快速定位某天是否异常
- 加一个“心跳检查”——每天跑一个独立脚本统计“昨天的数据记录数是否大于某个阈值”,如果低于阈值就输出告警信息
第三点实际上是数据质量监控的雏形。在个人项目阶段,哪怕只是每天手动看一眼数据量,也比完全不检查强得多。
5.3 站点编号变化导致数据断层
有一次我发现某站点的数据在某个时间点之后突然消失,排查后发现是平台升级,站点的station_id从纯数字变成了字母加数字的组合。这导致我原本按station_id关联的分析逻辑全部失效。
这个问题没法完全避免,但可以减轻影响:每次采集时把source字段记录下来,同时在存储层保留station_name,即使ID变了,还能通过站点名称人工确认是同一个站点。另外,定期对“站点列表”做一次全量快照,对比新旧ID的映射关系,可以尽早发现问题。
6. 从采集到分析的避坑心得与扩展方向
6.1 我踩过的最深的坑:单一数据源依赖
这个项目做到第三周时,平台的接口结构突然改版,原来的JSON字段全部变了名。我的爬虫一连跑了三天,解析出来的全是空值,但请求本身又是成功的——因为接口返回200,只是内容结构变了。
这个问题的教训非常深刻:任何数据采集项目都要有“源数据存档”的机制。我的做法是在解析之前,先把原始JSON完整保存一份到独立的目录,按日期命名。这样即使字段结构变了,历史数据还在,不会因为解析逻辑失效而丢失源头数据。
6.2 项目后续还能怎么扩展
公共自行车数据这个项目做完基础版之后,可以延伸的方向非常多。如果对机器学习感兴趣,可以用历史数据训练“站点车辆数预测模型”,输入是时间、天气、是否是工作日,输出是未来一小时的可用车辆数。这个课题在学术界和工业界都很成熟,但自己动手从采集到建模跑一遍,理解会深很多。
如果对可视化感兴趣,可以做实时监测大屏——每5分钟拉取一次数据,用WebSocket推送到前端页面,地图上的站点颜色随饱和度实时变化。整个链路涉及爬虫、消息队列、WebSocket、前端可视化,是一个完整的小型全栈项目。
如果对系统设计感兴趣,可以考虑把单机脚本改造成多城市并行采集:用concurrent.futures或者asyncio并发请求多个城市的接口,再做一个统一的数据存储层。这样数据量会从几个站点扩展到几百个站点,分析维度也会更丰富。
6.3 最后再分享一个小技巧
采集脚本运行一段时间后,数据库文件会越来越大。SQLite是单文件数据库,文件膨胀到一定程度后查询性能会下降。我的处理方式是每个月做一次数据归档:把上一月的数据从SQLite导出到CSV并压缩存档,只保留最近两个月的数据在SQLite中用于常规分析。这样既控制了库文件大小,也保留了完整的历史数据可供回看。
再到后来,我发现其实连“定时任务每次都全量采集”都不是最优解。公共自行车站点的基础信息(站点名、经纬度、总容量)变化频率很低,可以单独建一张station_info表,每天只更新一次;而车辆数、空桩数这类动态数据才需要高频率采集。把静态数据和动态数据分开存储,既能减少存储占用,也能让分析时的关联逻辑更清晰。这个思路在任何数据采集项目中都通用——先分清哪些是慢变量,哪些是快变量,再决定各自的采集频率。