前段时间，有个刚学 Python 的朋友问我：数据采集、数据清洗、数据分析，到底应该先学哪个？
我的回答是，别拆开学。
单独看教程，很容易变成这种状态：requests 会写两行，pandas 也见过，轮到自己做点东西，还是不知道从哪儿下手。
比较省事的办法，是找一份公开数据，从采集开始，一路做到分析结果。哪怕数据只有一百条，也比背一堆函数有用。
这篇就做一件小事：采集公开的文章数据，清洗掉不方便统计的内容，再看看每个用户发布了多少篇文章。
先把环境装好
电脑上需要有 Python 3。打开终端，输入：
python --version
能看到版本号，就继续安装两个库：
pip install requests pandas
requests 负责取数据，pandas 负责整理数据。新手第一遍不用装太多东西，库越多，报错时越难判断是哪一步的问题。
先看一眼数据从哪儿来
这次用的是 JSONPlaceholder。它是一个供开发和学习使用的公开测试接口，官方指南提供了 /posts 资源，可以直接返回文章列表。
浏览器打开下面这个地址，也能看到原始数据：
https://jsonplaceholder.typicode.com/posts
返回的数据大概是这样：
{
"userId": 1,
"id": 1,
"title": "文章标题",
"body": "文章正文"
}
一条记录有四个字段。我们要做的，就是把这些记录取回来，整理成表格。
第一步，采集数据
新建一个 main.py 文件，先写下面这段：
import requests
import pandas as pd
url = "https://jsonplaceholder.typicode.com/posts"
response = requests.get(url, timeout=10)
response.raise_for_status()
data = response.json()
df = pd.DataFrame(data)
print(df.head())
print(f"一共采集到 {len(df)} 条数据")
运行：
python main.py
如果终端里出现前五条记录，说明采集已经成功。
这里有两个细节我一般不会省。
一个是 timeout=10。网络一直没响应时，程序最多等十秒，不会挂在那里。
另一个是 raise_for_status()。如果接口返回 404、500 之类的错误，程序会直接报出来。否则有时请求明明失败了，后面的代码还在继续跑，排查起来挺绕。
先把原始数据保存下来：
df.to_csv(
"raw_posts.csv",
index=False,
encoding="utf-8-sig"
)
我做采集项目时，原始文件通常不改。清洗错了，还能回来重新处理。要是直接覆盖原始数据，后面发现误删，只能再采一次。
第二步，清洗数据
这份测试数据本身不算脏，不过很适合拿来练几个常用动作：
- 检查空值；
- 删除重复记录；
- 清理文本里的换行和多余空格；
- 增加后面分析要用的字段。
接着写：
clean_df = df.copy()
print("各字段空值数量：")
print(clean_df.isnull().sum())
clean_df = clean_df.dropna(
subset=["userId", "id", "title", "body"]
)
clean_df = clean_df.drop_duplicates(
subset=["id"]
)
clean_df["title"] = (
clean_df["title"]
.str.replace(r"\s+", " ", regex=True)
.str.strip()
)
clean_df["body"] = (
clean_df["body"]
.str.replace(r"\s+", " ", regex=True)
.str.strip()
)
clean_df["title_length"] = (
clean_df["title"].str.len()
)
clean_df["body_length"] = (
clean_df["body"].str.len()
)
clean_df.to_csv(
"clean_posts.csv",
index=False,
encoding="utf-8-sig"
)
有些人清洗数据时，一上来就是 dropna()，把带空值的整行全删了。这个习惯不太好。
假设一条文章记录只有正文为空，标题、作者和发布时间都正常，这条数据未必完全没用。删不删，应该看空的是哪个字段，以及它会不会影响当前分析。
这次四个字段都要用，所以才统一过滤。
去重也是一样。这里用 id，因为它是每篇文章的唯一编号。实际做商品采集时，可能要用商品编号；做新闻采集时，也可能要用文章链接加发布时间。不能看到“去重”两个字，就随手把整行重复删掉。
第三步，开始分析
现在来回答几个实际问题：
- 一共有多少个用户？
- 每个用户发布了多少篇文章？
- 哪个用户的平均标题更长？
继续补上代码：
user_count = clean_df["userId"].nunique()
post_count = (
clean_df.groupby("userId")
.size()
.reset_index(name="post_count")
)
avg_title_length = (
clean_df.groupby("userId")["title_length"]
.mean()
.round(1)
.reset_index(name="avg_title_length")
)
result = post_count.merge(
avg_title_length,
on="userId"
)
result = result.sort_values(
by=["post_count", "avg_title_length"],
ascending=[False, False]
)
print(f"用户数量：{user_count}")
print(result)
result.to_csv(
"user_analysis.csv",
index=False,
encoding="utf-8-sig"
)
groupby("userId") 可以理解成先把同一个用户的数据放到一组里，再分别统计。
size() 算的是每组有多少行，也就是每个用户有多少篇文章。
后面的 mean() 则是在算每组标题长度的平均值。最后用 merge() 把两份结果拼到一起。
这几行代码看着不多，但已经是一个完整的分析动作了：确定统计对象，选择指标，按用户分组，再合并结果。
完整代码
如果不想分段复制，可以直接运行下面这版：
import requests
import pandas as pd
def collect_data():
url = "https://jsonplaceholder.typicode.com/posts"
response = requests.get(url, timeout=10)
response.raise_for_status()
return pd.DataFrame(response.json())
def clean_data(df):
clean_df = df.copy()
clean_df = clean_df.dropna(
subset=["userId", "id", "title", "body"]
)
clean_df = clean_df.drop_duplicates(
subset=["id"]
)
for column in ["title", "body"]:
clean_df[column] = (
clean_df[column]
.str.replace(r"\s+", " ", regex=True)
.str.strip()
)
clean_df["title_length"] = (
clean_df["title"].str.len()
)
clean_df["body_length"] = (
clean_df["body"].str.len()
)
return clean_df
def analyze_data(df):
post_count = (
df.groupby("userId")
.size()
.reset_index(name="post_count")
)
avg_title_length = (
df.groupby("userId")["title_length"]
.mean()
.round(1)
.reset_index(name="avg_title_length")
)
result = post_count.merge(
avg_title_length,
on="userId"
)
return result.sort_values(
by=["post_count", "avg_title_length"],
ascending=[False, False]
)
def main():
raw_df = collect_data()
raw_df.to_csv(
"raw_posts.csv",
index=False,
encoding="utf-8-sig"
)
clean_df = clean_data(raw_df)
clean_df.to_csv(
"clean_posts.csv",
index=False,
encoding="utf-8-sig"
)
result = analyze_data(clean_df)
result.to_csv(
"user_analysis.csv",
index=False,
encoding="utf-8-sig"
)
print(f"原始数据：{len(raw_df)} 条")
print(f"清洗后数据：{len(clean_df)} 条")
print(result)
if __name__ == "__main__":
main()
别急着上爬虫框架
这套代码跑完，你手里会有三个文件：
raw_posts.csv：接口返回的原始数据；clean_posts.csv：去重并整理过的数据；user_analysis.csv：最终统计结果。
这已经把采集、清洗和分析串起来了。
后面可以把数据源换成网页，也可以增加图表，但我不建议刚跑通就开始堆多线程、代理池和浏览器自动化。新手最容易踩的坑，不是速度慢，而是采到的数据根本没法用。
先学会保留原始数据，给记录找准唯一标识，再想清楚最后要统计什么。这个习惯养成之后，换一个数据源，无非是采集部分的代码变了，后面的处理思路基本还能接着用。
正式采集其他网站时，也别忘了先查看网站条款和访问要求，只处理公开或已经获得授权的数据，不绕过登录、验证码和明确的访问限制。
