19_Python数据处理实战:CSV、JSON 与表格数据分析
Python数据处理实战:CSV、JSON 与表格数据分析
前面学了一堆语法和库,现在是把它们串起来的时候了。真实世界的编程,很多时候不是在"写算法",而是在处理数据——从某个地方拿到原始数据,清洗整理、计算统计、生成报告、导出给下一个环节使用。
这一篇以两个最通用的数据格式(CSV 和 JSON)为主线,带你走一遍从"拿到原始数据"到"产出分析报告"的完整流程。我们会用到标准库的 csv 和 json 模块,也会用到 pandas 来提升效率。
一、CSV 与 JSON
CSV 数据处理
读取 CSV
import csv
# 方法一:基本读取
with open("students.csv", "r", encoding="utf-8") as f:
reader = csv.reader(f)
header = next(reader) # 跳过表头
for row in reader:
print(row) # row 是列表
# 方法二:DictReader(每行是一个字典,用表头做键)
with open("students.csv", "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
print(f"{row['姓名']}:{row['成绩']}分")
写入 CSV
with open("output.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow(["姓名", "年龄", "成绩"]) # 表头
writer.writerows([
["小明", 20, 85],
["小红", 21, 92],
])
# DictWriter:用字典写入
with open("output.csv", "w", newline="", encoding="utf-8-sig") as f:
fieldnames = ["姓名", "年龄", "成绩"]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({"姓名": "小明", "年龄": 20, "成绩": 85})
newline="" 防止 Windows 上出现多余空行。encoding="utf-8-sig" 让 Excel 能正确打开中文 CSV。
实战:成绩数据清洗和分析
假设有这样一份原始数据,包含格式问题:
姓名,年龄,成绩
小明,20,85
小红,21,
小刚,20,78
小美,,92
小李,19,105
小张,18,-10
import csv
def clean_and_analyze(input_file, output_file):
valid_records = []
issues = []
with open(input_file, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for line_no, row in enumerate(reader, start=2):
name = row.get("姓名", "").strip()
age_str = row.get("年龄", "").strip()
score_str = row.get("成绩", "").strip()
# 校验逻辑
errors = []
if not name:
errors.append("姓名为空")
try:
age = int(age_str) if age_str else None
except ValueError:
errors.append(f"年龄 '{age_str}' 不是有效整数")
age = None
try:
score = float(score_str) if score_str else None
except ValueError:
errors.append(f"成绩 '{score_str}' 不是有效数字")
score = None
if score is not None and not (0 <= score <= 100):
errors.append(f"成绩 {score} 超出 0~100 范围")
if errors:
issues.append({"行号": line_no, "姓名": name, "问题": "; ".join(errors)})
else:
valid_records.append({"姓名": name, "年龄": age, "成绩": score})
# 统计分析
if valid_records:
scores = [r["成绩"] for r in valid_records if r["成绩"] is not None]
ages = [r["年龄"] for r in valid_records if r["年龄"] is not None]
print(f"有效记录:{len(valid_records)} 条")
print(f"问题记录:{len(issues)} 条")
if scores:
print(f"平均分:{sum(scores)/len(scores):.1f}")
print(f"最高分:{max(scores)}")
print(f"最低分:{min(scores)}")
print(f"及格率:{sum(1 for s in scores if s>=60)/len(scores)*100:.1f}%")
# 输出问题报告
if issues:
print(f"\n数据问题清单:")
for issue in issues:
print(f" 第 {issue['行号']} 行 ({issue['姓名']}):{issue['问题']}")
# 写入清洗后的数据
if valid_records:
with open(output_file, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["姓名", "年龄", "成绩"])
writer.writeheader()
writer.writerows(valid_records)
print(f"\n清洗后的数据已保存到 {output_file}")
# 创建测试数据并运行
with open("raw_students.csv", "w", encoding="utf-8") as f:
f.write("姓名,年龄,成绩\n小明,20,85\n小红,21,\n小刚,20,78\n小美,,92\n小李,19,105\n小张,18,-10\n")
clean_and_analyze("raw_students.csv", "clean_students.csv")
JSON 数据处理
读取和解析 JSON
import json
# 从文件读取
with open("config.json", "r", encoding="utf-8") as f:
config = json.load(f)
# 从字符串解析
json_str = '{"name": "小明", "scores": [85, 92, 78]}'
data = json.loads(json_str)
print(data["scores"][1]) # 92
写入 JSON
data = {"app": "MyApp", "version": "1.0", "debug": False}
# 写入文件
with open("config.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 转为字符串
text = json.dumps(data, ensure_ascii=False, indent=2)
处理复杂的嵌套结构
# 处理嵌套的 JSON 数据(类似 API 返回的数据)
response = {
"status": "success",
"data": {
"students": [
{"id": "001", "name": "小明", "scores": {"math": 85, "english": 90}},
{"id": "002", "name": "小红", "scores": {"math": 92, "english": 88}},
]
}
}
for student in response["data"]["students"]:
avg = sum(student["scores"].values()) / len(student["scores"])
print(f"{student['name']} 平均分:{avg:.1f}")
# 安全访问嵌套字段(防止 KeyError)
student = response.get("data", {}).get("students", [{}])[0]
name = student.get("name", "未知")
print(name)
二、pandas 数据分析
pandas 速成——从 CSV 到统计图表只需 5 分钟
pandas 的功能太多了——官方文档有几万页。但 80% 的日常数据处理工作只用到它的 20% 的功能。下面是最常用的这 20%,集中展示了从读文件到出统计的完整流程:
import pandas as pd
# 1. 读取
df = pd.read_csv("sales.csv", encoding="utf-8")
# 2. 看一眼
print(df.head())
print(df.info())
print(df.describe()) # count, mean, std, min, 25%, 50%, 75%, max
# 3. 筛选
high_value = df[df["金额"] > 1000]
beijing_orders = df.query("city == '北京' and amount > 500")
# 4. 排序
df_sorted = df.sort_values("金额", ascending=False)
# 5. 分组聚合
by_city = df.groupby("城市").agg(
总金额=("金额", "sum"),
平均金额=("金额", "mean"),
订单数=("金额", "count")
).round(2)
# 6. 新增列
df["折扣价"] = df["金额"] * 0.9
df["等级"] = pd.cut(df["金额"], bins=[0, 100, 500, 1000, 99999], labels=["小", "中", "大", "特大"])
# 7. 处理缺失
df["备注"] = df["备注"].fillna("无")
df = df.dropna(subset=["金额"]) # 删除金额为空的行
# 8. 导出
df.to_csv("cleaned.csv", index=False, encoding="utf-8-sig")
df.to_excel("report.xlsx", index=False)
把这几个操作练熟,你就能独立处理大多数表格数据任务。其余的功能(时间序列、透视表、数据合并……)用到的时候查文档,一次就能学会。
CSV 和字符编码——一个让你的 Excel 用户感激你的细节
如果你用 encoding="utf-8" 保存 CSV,在 Excel 中直接打开,中文可能会显示为乱码。这是因为 Excel 在打开 CSV 时不读取文件的编码声明,而是根据系统区域设置猜测编码。中文 Windows 默认猜测 GBK。
解决方案是用 encoding="utf-8-sig"。utf-8-sig 和 utf-8 的唯一区别是:前者在文件开头加了一个 BOM(字节顺序标记,)。这个 BOM 对 Python 和其他编程语言的读取没有影响(它们会自动忽略),但对 Excel 来说是"嘿,我是 UTF-8!"的信号。用 utf-8-sig 编码保存的 CSV 在 Excel 中打开时中文不会乱码。
另外,newline="" 在 Windows 上也很重要——不传这个参数的话,csv.writer 在 Windows 上写出的每行之间会多一个空行。
用 pandas 加速数据处理
CSV 和 JSON 的标准库模块足够用,但当数据量大、操作复杂时,pandas 的效率是数量级的提升:
import pandas as pd
# 读取数据
df = pd.read_csv("students.csv", encoding="utf-8")
# 数据概览
print(df.head()) # 前 5 行
print(df.info()) # 列信息和数据类型
print(df.describe()) # 数值列的统计摘要
# 筛选
high_scores = df[df["成绩"] >= 90]
young_students = df.query("年龄 <= 20 and 成绩 >= 80")
# 分组统计
class_stats = df.groupby("班级").agg({
"成绩": ["mean", "max", "min", "count"],
"姓名": "count"
}).round(1)
print(class_stats)
# 排序
df_sorted = df.sort_values("成绩", ascending=False)
# 新增计算列
df["等级"] = pd.cut(df["成绩"], bins=[0, 60, 70, 80, 90, 100],
labels=["F", "D", "C", "B", "A"])
# 导出
df.to_csv("processed.csv", index=False, encoding="utf-8-sig")
df.to_excel("processed.xlsx", index=False)
df.to_json("processed.json", orient="records", force_ascii=False)
处理缺失值
# 检查缺失值
print(df.isnull().sum())
# 删除有缺失的行
df_clean = df.dropna()
# 填充缺失值
df["成绩"] = df["成绩"].fillna(df["成绩"].mean())
df["年龄"] = df["年龄"].fillna(0)
三、实战项目
实战:完整的天气数据分析报告
import json
import csv
from datetime import datetime, timedelta
from collections import defaultdict
def analyze_weather_data(json_file):
"""分析天气 JSON 数据,生成报告"""
with open(json_file, "r", encoding="utf-8") as f:
data = json.load(f)
records = data.get("daily", [])
if not records:
print("无天气数据")
return
# 统计
temps_high = []
temps_low = []
conditions = defaultdict(int)
for day in records:
date = day.get("date", "未知")
high = day.get("temp_high")
low = day.get("temp_low")
condition = day.get("condition", "未知")
if high is not None:
temps_high.append((date, high))
if low is not None:
temps_low.append((date, low))
conditions[condition] += 1
# 生成报告
print("===== 天气数据分析 =====")
print(f"统计天数:{len(records)}")
if temps_high:
hottest = max(temps_high, key=lambda x: x[1])
coldest = min(temps_low, key=lambda x: x[1])
avg_high = sum(h for _, h in temps_high) / len(temps_high)
avg_low = sum(l for _, l in temps_low) / len(temps_low)
print(f"平均最高温:{avg_high:.1f}°C")
print(f"平均最低温:{avg_low:.1f}°C")
print(f"最热的一天:{hottest[0]} ({hottest[1]}°C)")
print(f"最冷的一天:{coldest[0]} ({coldest[1]}°C)")
print(f"\n天气分布:")
for cond, count in sorted(conditions.items(), key=lambda x: -x[1]):
bar = "█" * count
print(f" {cond}: {bar} {count} 天")

图19-1 CSV 与 JSON 数据格式对比:CSV 适合表格数据(简单、Excel 兼容),JSON 适合嵌套结构(灵活、API 标准),各有最佳使用场景。
真实世界的数据长什么样——脏、乱、差
教科书里的数据总是一行不缺、格式完美、类型正确。真实世界的数据完全不是这样。从网上下载的 CSV 文件可能有中英文混用的表头、某一行突然少了一个字段、日期列里夹着文字备注、“未知"和"保密"和空着三种方式表示"没有填写”。从 API 拿到的 JSON 可能有深层嵌套、字段名拼写不一致、空数据用 null 和 "" 和 [] 三种方式表示。
数据清洗不是"锦上添花",而是数据处理的第一步。而且往往是耗时最长的一步。业界有句半开玩笑的话:数据科学家 80% 的时间在清洗数据,20% 的时间在抱怨为什么数据这么脏。本文中 clean_and_analyze() 函数的校验逻辑——检查空值、检查数据类型、检查取值范围——就是数据清洗的微缩版。虽然看起来简单,但这个"读取 → 逐行校验 → 过滤无效 → 统计有效"的模式,是几乎所有真实数据处理管道的起点。
选 CSV 还是 JSON——一个实用的决策框架
CSV 和 JSON 是两种最通用的数据交换格式,选哪个取决于你的场景。CSV 适合:行列整齐的表格数据、需要 Excel 打开查看或编辑、接收方只支持 CSV 格式。JSON 适合:数据有嵌套层级(比如学生下面有各科成绩)、需要在不同编程语言之间交换数据、API 调用返回的数据。一个简单的判断方式:如果你能用一个 Excel 表格清晰地表示你的数据结构,就用 CSV。如果不能——比如一个单元格里要放一个列表——就用 JSON。
在实际项目中,两种格式往往混合使用:你可能从一个 API 拿到 JSON 格式的原始数据,解析后提取有用的字段,处理成表格形式,最后导出为 CSV 方便业务人员用 Excel 查看。这个"JSON → Python 处理 → CSV"的转换流程,是数据分析中最常见的工作模式之一。

图19-2 数据处理管道:原始数据 → 读取 → 校验清洗 → 转换计算 → 统计分析 → 导出结果,五步管道流程。
实战:JSON 数据深度提取——处理 API 返回的复杂嵌套结构
真实世界的 API 返回的 JSON 往往有 3-5 层嵌套。直接写 data["a"]["b"]["c"] 会在任何一层缺失时崩溃。一个实用的模式是用"安全路径"来访问深层字段:
def safe_get(data, *keys, default=None):
"""安全访问嵌套字典/列表,任意一层缺失都返回 default"""
try:
for key in keys:
if isinstance(data, list):
key = int(key)
data = data[key]
return data
except (KeyError, IndexError, TypeError, ValueError):
return default
# 模拟一个深度嵌套的 API 响应
response = {
"data": {
"users": [
{"profile": {"name": "小明", "contact": {"email": "xiao@mail.com"}}},
{"profile": {"name": "小红"}} # 没有 contact 字段
]
}
}
for user in response.get("data", {}).get("users", []):
name = safe_get(user, "profile", "name", default="未知")
email = safe_get(user, "profile", "contact", "email", default="未填写")
print(f"{name}: {email}")
# 小明: xiao@mail.com
# 小红: 未填写 ← 不会崩溃!
这个 safe_get 函数在处理外部数据时是救命的——你永远不知道 API 返回的数据在哪个嵌套层级出现字段缺失。把它放进你的 utils.py 工具箱里,以后每次处理嵌套 JSON 时都会用到它。
四、流程与总结
什么是"整洁数据"——数据处理中的一个关键概念
在数据处理领域有一个著名的原则叫"整洁数据"(Tidy Data),由统计学家 Hadley Wickham 提出。它的核心规则只有三条:每一列是一个变量、每一行是一个观测、每个观测单元组成一张表。听起来简单,但真实世界百分之八十的数据都不满足这个标准——比如有些 CSV 文件把月份作为列名(一月销售额、二月销售额……),而不是把"月份"作为一个变量列。
遇到"不整洁"的数据时,不需要写复杂的逻辑来适配——让 pandas 帮你把数据整理成标准形式。pd.melt() 把宽表变长表(列名变成行),pd.pivot_table() 做数据透视。这些操作初学时不常用到,但了解"整洁数据"这个概念,在你未来面对杂乱数据时能提供一个清晰的思考框架:我现在的数据哪里不整洁?把它整理整洁之后,所有的分析操作都会变得统一和简单。
pandas 的强大之处就在于它围绕"整洁数据"的理念设计了一整套 API——你先把数据整理成标准形态,然后一切操作(筛选、分组、聚合、绘图)都变得可预测和一致。花 10 分钟在搜索引擎里搜一下"Tidy Data"这个概念,它会改变你对数据处理的理解。
数据处理的通用流程
不管你处理什么数据,以下五个步骤几乎每次都出现:
- 读取:从文件、API、数据库获取原始数据
- 清洗:处理缺失值、异常值、格式不一致、重复数据
- 转换:类型转换、单位换算、计算衍生字段
- 分析:聚合、统计、分组对比、趋势发现
- 输出:生成新文件、打印报告、可视化图表
这五个步骤形成了一条数据处理管道。你写的每个数据处理程序,本质上都是这条管道的某种组合。
实战:一个完整的"电商订单数据清洗与分析"管线
把 CSV 处理、数据清洗、统计分析和结果输出的完整流程串起来,写一个模拟电商订单分析的脚本:
import csv
import json
from collections import defaultdict
from pathlib import Path
# 模拟原始订单数据(实际场景中可能来自 CSV 文件或 API)
raw_orders = [
{"order_id": "O001", "customer": "小明", "product": "iPhone 15", "amount": "8999", "status": "completed", "date": "2025-07-01"},
{"order_id": "O002", "customer": "小红", "product": "AirPods", "amount": "", "status": "pending", "date": "2025-07-02"},
{"order_id": "O003", "customer": "小明", "product": "iPad", "amount": "3499", "status": "completed", "date": "2025-07-02"},
{"order_id": "O004", "customer": "小李", "product": "MacBook", "amount": "14999", "status": "cancelled", "date": "2025-07-03"},
{"order_id": "O005", "customer": "小红", "product": "Apple Watch", "amount": "2999", "status": "completed", "date": "2025-07-04"},
{"order_id": "O006", "customer": "", "product": "iPhone 15", "amount": "8999", "status": "completed", "date": "2025-07-05"},
]
# 第一步:数据清洗
clean_orders = []
issues = []
for i, row in enumerate(raw_orders):
errors = []
if not row.get("customer", "").strip():
errors.append("客户名为空")
try:
amount = float(row.get("amount", 0)) if row.get("amount") else 0
if amount <= 0:
errors.append(f"金额异常:{amount}")
except (ValueError, TypeError):
errors.append(f"金额格式错误:{row.get('amount')}")
amount = 0
if errors:
issues.append({"行号": i+2, "订单号": row.get("order_id"), "问题": "; ".join(errors)})
continue
clean_orders.append({**row, "amount": amount})
# 第二步:统计分析
total_revenue = sum(o["amount"] for o in clean_orders)
avg_order = total_revenue / len(clean_orders) if clean_orders else 0
by_customer = defaultdict(lambda: {"count": 0, "total": 0})
by_product = defaultdict(int)
by_status = defaultdict(int)
for o in clean_orders:
by_customer[o["customer"]]["count"] += 1
by_customer[o["customer"]]["total"] += o["amount"]
by_product[o["product"]] += 1
by_status[o["status"]] += 1
# 第三步:输出报告
print("===== 电商订单分析报告 =====")
print(f"有效订单:{len(clean_orders)} 条")
print(f"问题订单:{len(issues)} 条")
print(f"总营收:¥{total_revenue:,.2f}")
print(f"客单价:¥{avg_order:,.2f}")
print()
print("客户消费排名:")
for name, data in sorted(by_customer.items(), key=lambda x: -x[1]["total"]):
print(f" {name}: {data['count']}笔, ¥{data['total']:,.2f}")
print()
print("商品热度排名:")
for product, count in sorted(by_product.items(), key=lambda x: -x[1]):
print(f" {product}: {count}件")
print()
print("订单状态分布:")
for status, count in by_status.items():
print(f" {status}: {count}单")
if issues:
print(f"\n⚠ 数据质量问题({len(issues)} 条):")
for issue in issues:
print(f" 第{issue['行号']}行 [{issue['订单号']}]: {issue['问题']}")
# 第四步:导出清洗后的数据
output = Path("clean_orders.json")
output.write_text(json.dumps(clean_orders, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"\n清洗后的数据已保存到 {output}")
这个脚本充分体现了数据处理管道的五个阶段:读取 → 清洗(校验金额、客户名)→ 转换(字符串转数值)→ 分析(按客户/商品/状态分组聚合)→ 输出(JSON 文件 + 屏幕报告)。虽然只有 80 行,但它的结构可以直接放大到处理真实业务数据——保留这个管道结构,替换每一阶段的具体逻辑即可。
数据处理的心流——从原始数据到可读报告
如果你回顾本文的所有代码示例,会发现一个反复出现的模式:所有数据处理脚本都是"接收某种格式的原始数据 → 经过一系列转换 → 输出更有价值的信息"。无论是清洗 CSV、嵌套 JSON 提取、还是统计分析,底层结构惊人地一致。
掌握数据处理不是记住所有的 pandas 方法或正则表达式符号,而是建立起这种"管道思维":数据流经一系列处理阶段,每阶段做一件事。当你在解决一个新的数据处理问题时,第一步不是写代码——是在草稿纸或脑子里画出管道:输入是什么格式?经过哪几个转换步骤?输出是什么格式?画清楚管道之后,每个阶段单独实现、单独测试,最后串联。
你做了足够多的数据处理练习之后,会发现很多看起来完全不同的问题——“统计网站访问日志”“分析考试成绩”“整理电商订单”——在管道层面共享完全相同的结构。这种"看到不同问题的共同骨架"的能力,是经验丰富的开发者和新手的核心区别之一。
动手练习
-
写一个"API 数据管道":找一个免费的公共 API(如天气 API、汇率 API、COVID 数据 API),用 requests 获取 JSON 数据,解析嵌套结构提取你关心的字段,清洗并整理成 CSV 格式,用 pandas 做简单的统计分析(平均值、趋势、Top N),最后用 matplotlib 生成一张包含两个子图的报告(趋势图 + 分布图)。这个练习把 API 调用、JSON 解析、数据清洗、统计分析和数据可视化五个环节完整串联了起来——它是本文所有知识点的综合应用,也是日常工作场景的微缩版。
-
CSV 合并工具:写一个程序,读取一个目录下的所有 CSV 文件(结构相同),合并成一个大文件,添加一列标明每行来自哪个文件。
-
JSON 到 CSV 转换器:写一个函数,读取嵌套的 JSON 数据,展平成 CSV 格式。处理
list和dict的嵌套结构。 -
学生成绩综合分析:读取一个包含多科成绩的 CSV,计算每个学生的总分、平均分、排名,按班级统计各科均分,最后输出到 Excel。
-
改进电商订单分析脚本:在本文的电商订单脚本基础上,增加以下功能:按日期范围筛选、计算每日营收趋势、找出"下了单但取消率最高"的客户、把统计报告输出为格式化的文本文件而非只打印到屏幕。做完了你会发现:从"一个能跑的分析脚本"到"一个真正能給业务人员使用的数据分析工具",需要的不是更高级的算法,而是更好的输入校验、更清晰的输出格式、更全面的边界情况处理。这些"非算法"的能力,恰恰是工作中最被低估但也最被需要的技能的所在。一个好的数据分析脚本和一个差的之间,往往不是算法复杂度的差异,而是前者在输入输出上考虑周全——输入能够容忍用户的格式错误,输出能够让不懂技术的业务人员一眼看懂。能够在"技术正确"和"用户友好"之间找到平衡,是一个成熟的开发者区别于初级开发者的核心标志。这种能力不是从算法书里学来的,是从一次次被用户"不按预期使用"中磨出来的。当你把数据处理脚本交给同事使用时,ta 可能会传入一个中文表头的 CSV(你的脚本只处理英文表头),或者在金额栏里写了"约 500 元"(你的脚本期望纯数字),或者给了你一个 2GB 的文件(你的脚本试图一次性读入内存)。每一次"意料之外的输入",都是加固你脚本健壮性的机会。处理得多了,你自然就养成了"在任何函数入口处先做输入校验"的习惯——而这个习惯是写出工业级代码的核心素养。数据处理的世界里,"脏数据"不是例外,而是常态。能够优雅地处理脏数据、让分析在输入不完美时依然产生有意义的结果、在数据和预期不符时给出清晰的错误提示——这些能力比任何炫目的算法都更能决定一个数据工程师的职业高度。
要点回顾
- CSV 用
csv.reader/DictReader读取,csv.writer/DictWriter写入 - JSON 用
json.load/dump读写文件,json.loads/dumps读写字符串 - 数据处理五步:读取 → 清洗 → 转换 → 分析 → 输出
- 数据校验是数据处理的第一步——异常值和缺失值不处理,分析结果就不可靠
- pandas 在处理表格数据时比手写循环快得多、代码少得多
下篇预告
下一篇是最后一篇——综合项目实战,把 19 篇学到的所有知识融进一个完整的学生信息管理系统。
更多推荐
所有评论(0)