数据获取与处理实战：从数据源选择到数据清洗的完整指南

这几年做数据相关的工作，被问得最多的一个问题是：“我该从哪儿拿数据？拿到手之后又该怎么处理？”说实话，大多数人在第一步就卡住了，要么不知道怎么稳定地拿到数据，要么千辛万苦拿到一份Excel，结果里面全是乱码、重复行、格式不一致，光清洗就耗掉半天。我把这些年“踩坑攒经验”的过程整理了一下，这篇东西不聊虚的，全是数据处理与获取过程中真正能落地的技巧：怎么选数据源、怎么写一个不容易崩的请求函数、怎么把脏数据洗得干干净净、怎么存才能方便后续使用，以及我在实际项目里遇见的各种奇葩问题和排错思路。适合准备入行的数据分析师、刚接手数据采集任务的开发，以及所有被“脏数据”折磨过的人。

1. 数据获取的先决条件：先想清楚来源，再动手写代码

很多新手拿到需求的第一反应是“打开网页、右键查看源代码、开爬”。但这样做十次有八次要返工，因为数据获取不是“写个请求”这么简单，你选的数据源直接决定了后续所有环节的工作量。所以真正开始前，我建议先花时间做一个判断：数据到底要以什么方式获取，是最优先的。

1.1 三类常见数据源，怎么选最省力

我平时把数据源分成三类：

第一类：开放API接口。这是最理想的情况。平台方已经封装好了数据输出格式，你只需要看文档、申请权限、按照参数调用即可。数据结构稳定，更新及时，也不容易触发访问限制。缺点是接口往往有频率限制、需要Token、有些数据要付费或者申请审核。

第二类：静态HTML页面。很多行业网站、公开统计页面、政府公示页数据都在HTML里，不涉及动态加载。这种页面直接用HTTP库请求，然后解析HTML拿数据，过程比较简单粗暴。风险点在于页面改版会导致解析代码失效，所以一定要把解析逻辑封装好，方便维护。

第三类：动态渲染页面。比如用React、Vue做的网站，数据不是直接写在HTML源代码里的，而是页面加载后通过JavaScript发请求再渲染。这时候千万别一上来就搞无头浏览器，先按下F12，打开开发者工具里的“网络”面板，刷新页面，找XHR或Fetch请求。数据绝大多数情况都在某个JSON接口里，直接请求那个接口，速度比无头浏览器快得多，成功率也高。

举一个真实遇到过的例子。有一次我要抓某个行业指数网站的历史数据，直接用HTTP请求首页，发现返回的HTML里全是空的div。打开网络面板一看，发现页面加载后会请求一个带日期参数的JSON文件，返回就是完整的时间序列。我直接把请求地址里的日期参数改掉，绕开了所有的DOM解析，一分钟就拿到了十年的数据。这就是为什么我总说，动手之前先打开控制台看看流量走了哪些请求，远比自己瞎猜要靠谱。

1.2 请求工程的基础操作：超时、重试、请求头都要认真对待

很多人写请求代码翻车，不是不会requests，而是没把“工程化”这件事做好。所谓的“工程化”，体现在几个不起眼的小地方：

请求头要构造像真实用户。最基本的User-Agent一定要设置。有些网站还会检查Referer，意思就是看你这个请求是从哪个页面过去的。模板可以这样：

import requests from fake_useragent import UserAgent HEADERS = { "User-Agent": UserAgent().random, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", }

需要传Referer时继续往里面加。注意fake_useragent这个库偶尔会命中反爬策略里的“非浏览器特征”，所以更稳妥的做法是准备一个固定的、常用的浏览器UA字符串，而不是每次都随机。

超时和重试必须具备。网络请求不设置超时，程序会一直挂在那里，尤其在批量抓取时，一个卡死的请求能把整个流程堵住。我自己习惯用元组形式同时设置连接超时和读取超时：

resp = requests.get(url, headers=HEADERS, timeout=(3, 10))

如果请求失败，比如连接被重置、超时、返回5xx，不能直接就放弃，建议加一个带退避的重试逻辑：

import time import requests from requests.adapters import HTTPAdapter s = requests.Session() s.mount("https://", HTTPAdapter(max_retries=3)) def fetch(url, retries=3): for i in range(retries): try: resp = s.get(url, headers=HEADERS, timeout=(3, 10)) if resp.status_code in (200, 404): return resp if resp.status_code in (429, 500, 502, 503): time.sleep(2 * (i + 1)) continue except requests.RequestException as e: time.sleep(2 * (i + 1)) return None

这里面的逻辑是：404说明资源本身不存在，没必要重复请求；429/5xx属于临时性问题，退避重试一下大概率能过。

控制请求频率。不要一口气发几百个请求，特别是对没有你账号体系的公开网站。合规的获取方式应当遵守数据源方的使用条款与限速要求，这是前提。如果确实需要大量数据，先把请求间隔控制了，比如每请求一次sleep 0.5到1秒。慢是慢一点，但胜在稳定。

提示：数据获取必须遵守目标网站的服务条款与robots规则，仅获取公开信息，并且在用于商业用途前确认数据授权方式。很多纠纷都是因为过度请求或滥用数据导致的，这一条值得给所有做数据的人提个醒。

2. 从网页和接口中精准抽取目标数据

数据源一旦确定，接下来就是“解析”。这块最忌讳的就是把解析逻辑写成一次性脚本，只跑一次就完事。只要你的数据需要定期更新，就应该把解析代码写得足够灵活，这样哪怕页面结构微微变化，你也只需要改一小段配置而不是重写。

2.1 HTML解析的工具选型与组合策略

解析HTML有哪些选项？正则表达式、XPath、CSS选择器、BeautifulSoup、lxml，等等。我的实践经验是这样：

正则表达式适合处理非常规整、短小且没有嵌套的文本。比如从一段JavaScript代码里提取json字符串、从页面中提取邮箱或电话号码。但它不适合解析整个HTML文档，因为HTML的标签嵌套太深，正则根本处理不了层级关系，硬写只会把自己绕晕。

BeautifulSoup + lxml 是新手最顺手的组合。尤其是find、find_all配合CSS选择器，写起来很直观。但因为BeautifulSoup对节点操作非常灵活，大型页面上性能不算最好。如果抓的是列表页还好，一旦遇上那种几万行的大表格，解析速度会明显变慢。

lxml直接配合XPath，是我目前的主力方案。XPath能精准地表达“我要第几个表格里的第几行”这种复杂逻辑，而且底层是C语言实现，速度很快。一个典型的写法：

from lxml import html tree = html.fromstring(page_source) # 提取所有tr节点 for tr in tree.xpath('//table[@id="data_list"]//tr'): cells = tr.xpath('./td/text()') if not cells: continue record = [cell.strip() for cell in cells] print(record)

为什么用lxml而不是正则？因为数据抽取最重要的是“定位”，而不是“匹配文本”。XPath让人能直接依赖标签层级和属性进行定位，比如//tr[contains(@class, 'item')]这种写法，正则写起来会非常痛苦。

2.2 JSON接口返回数据怎么处理

现在越来越多的数据源直接返回JSON，这比解析HTML要省力得多。但JSON也有自己的问题：嵌套特别深、字段名不统一、数组结构模拟两可。处理这块我常用的套路是：

• 先用json.loads把字符串变成字典。
• 再用jsonpath-ng或者标准的递归遍历去抽取目标字段。
• 遇到缺失字段时，用.get方法指定默认值，不要直接抛异常。
• 如果要把嵌套JSON展平成表格，直接用pandas.json_normalize，它能把层级嵌套的列表、字典自动展开成二维表格。

举个例子：

import json from pandas import json_normalize raw = json.loads(resp.text) records = raw["data"]["list"] df = json_normalize(records, sep="_")

这里有个经验：接口返回的字段名通常比较“开发视角”，比如user_id、created_at，但在后续分析时我们可能希望它叫用户ID、创建时间。规范的流程是取到原始数据后立刻做一次字段重命名，形成一个标准化的schema，这样下游不管谁用，看到的名字都是稳定的。

2.3 表格抽取的隐蔽坑位

网页里最常见的结构化数据载体是HTML表格，但想从表格里提取出干净的数据，并不像看起来那么简单。我经常遇到这么几种情况：

表头不固定。有些页面第一行是标题，第二行才是数据表头；有些页面有合计行混在中间。写解析时不要用死位置，至少要加一个判断：这行的第一个单元格是否为“序号”“名称”之类的表头标识。

合并单元格。在展示层里，表格经常会把相同内容的相邻行合并成一个单元格，但HTML源码里可能只有一个td，这就导致下面的行缺列。处理方式是根据当前行的列数动态补齐，或者用前一行合并单元格里的值来填充当前行缺失的字段。

翻页接口位置不对。大部分翻页除了“下一页”按钮，还有一个隐藏的页码参数。建议不要用正则去抓“下一页”按钮的href，因为很多站点的翻页是JavaScript动态生成的事件，并没有真正的href属性。最好的办法仍然是打开网络面板，观察点击下一页时候产生了哪个新请求。

3. 数据清洗：把“能看”的数据变成“能用”的数据

我一直觉得，获取数据只是第一公里的活儿，数据清洗才真正决定你后续分析是否靠谱。很多情况下拿到的数据格式千疮百孔，根本原因是数据生产方自己的录入格式就不统一。清洗的本质不是写一长串代码，而是定义“什么叫干净的数据”，然后逐个把脏数据转化成你定义的标准形态。

3.1 脏数据的常见面谱与整体清洗流程

我在实际项目里见过的脏数据大概有这几类：

文本格式不统一。同一列里，有人填“1000元”，有人填“1,000”，有人填“一千”，还有人是全角数字。这种情况需要统一转半角，去掉货币符号和千分位分隔符，再识别人工中文数字。

日期格式混乱。同一个日期列可能出现“2023-01-01”“2023/1/1”“20230101”“2023年1月1日”等多种写法。可以用pandas的to_datetime配合format参数去解析，但纯靠函数解析不保证100%准确，最好先把日期统一转换成标准字符串格式。

空值处理。拿到数据后先看每列的空值比例。空值不是非得填充，要看业务含义。如果订单金额空，可能表示未支付或数据缺失，处理策略完全不同。

重复记录。特别是整合多个来源数据的时候，同一实体出现多次。重复的定义也要先想清楚：是主键完全相同才算重复，还是某些关键字段相同就算重复。

异常值。比如年龄出现200，销售额出现负数。异常值不能直接删，要留一份原始存档并记录删除原因，这样后续复盘的时候能说清楚数据为什么少了一截。

我的清洗流程基本固定为六步：备份原始数据 -> 定义字段类型与格式标准 -> 去重 -> 缺失值处理 -> 异常值识别 -> 输出清洗报告。

3.2 用Pandas高效处理常见清洗场景

Pandas是做数据清洗最顺手的工具，但很多人只会用str.strip()和fillna()。我分享几个实际使用频率很高的组合动作：

文本清理实战模板：

df["金额"] = ( df["金额"] .astype(str) .str.replace(",", "", regex=False) .str.extract(r"(-?\d+\.?\d*)")[0] .pipe(pd.to_numeric, errors="coerce") )

这个思路是先去掉千分位逗号，再用正则把数字部分提取出来，最后统一转成数值类型。注意extract返回的是DataFrame，所以取[0]才变成Series。

单位归一化：

有些数据里金额单位写成“万”，有些写成“亿”，分析时如果不统一就是灾难。我习惯这样处理：

def parse_money(text): if isinstance(text, str): if text.endswith("亿"): return float(text[:-1]) * 100000000 if text.endswith("万"): return float(text[:-1]) * 10000 return float(text) return text

这个简单逻辑能解决80%的单位问题。注意有些字符串里带空格，先strip()再判断。

日期解析：

df["日期"] = pd.to_datetime(df["原始日期"], format="%Y-%m-%d", errors="coerce")

这里必须提一句：errors="coerce"的意思是解析失败时填NaT（缺失时间）。加上这个参数不会让程序崩溃，但你还得回过头来看一下有多少行变成了NaT，如果比例太高说明原始格式判断错了。

重复值处理：

df = df.drop_duplicates(subset=["主键字段"], keep="first")

keep="first"表示保留第一次出现的记录。如果你的业务是每天跑增量，那么更合理的做法是保留最新一条，设置keep="last"。

3.3 编码错乱与“看不见的空格”怎么解

我一度以为乱码问题只存在于老旧的Windows系统，直到后来发现，哪怕数据源是网页，也可能因为服务端返回的编码声明与实际内容不一致导致乱码。解决办法很简单：

resp.encoding = resp.apparent_encoding

这行代码让requests用响应内容反推编码，虽然偶尔不准，但比默认的iso-8859-1要好太多。下载到本地文件后如果读取乱码，可以先用chardet检测：

import chardet with open("data.csv", "rb") as f: raw = f.read() encoding = chardet.detect(raw)["encoding"]

然后重新按检测到的编码读取。

另一种坑是“看不见的空格”。字符串里可能包含\u3000（全角空格）、\xa0（不间断空格）、零宽字符等。普通strip()处理不了，需要这样清：

df["文本"] = ( df["文本"] .str.replace("\u3000", "", regex=False) .str.replace("\xa0", "", regex=False) .str.strip() )

4. 数据存储与输出：别让Excel成为唯一归宿

数据清洗完不代表事情结束，怎么存、怎么交付，同样影响后续的使用体验。很多人习惯把结果直接扔给Excel，但有些场景下Excel并不是最优解。

4.1 不同存储场景的选型逻辑

我个人的选择标准是这样：

如果要交付给不懂技术的人看，优先Excel。但注意两点：一是写入Excel之前，把字段用英文或规范的拼音命名，避免一些老版本软件打开时报表乱码；二是用openpyxl或pandas.to_excel导出时，可以顺便设置列宽、冻结首行，让业务方一眼能看懂。

如果数据量几百万到几千万行，但并不需要多人同时写，用SQLite足够。SQLite是单文件数据库，不需要单独部署服务端，随取随用，尤其适合做本地数据处理过程中的中间层。

如果数据需要支撑一个在线分析系统或者Web应用，那就要用MySQL、PostgreSQL这类成熟的关系型数据库。写入时不要一条条insert，用批量写入，比如executemany或to_sql，一次性插入几千行。

贴一个Pandas直接写数据库的例子：

from sqlalchemy import create_engine engine = create_engine("sqlite:///output.db") df.to_sql("table_name", engine, if_exists="replace", index=False)

这里的if_exists="replace"是全表覆盖模式。做增量更新的时候更常用if_exists="append"，或者先查已有主键，只插入新增行。

如果只是给自己做后续分析和机器学习，直接存成Parquet。Parquet格式有压缩、列式存储，读取速度明显优于CSV。可以用三行代码搞定：

df.to_parquet("data.parquet", index=False) new_df = pd.read_parquet("data.parquet")

4.2 文件输出避坑：编码、大文件与增量更新

写CSV文件的时候，最经典的问题就是用Excel打开中文乱码。解决办法只有一个：输出的编码必须是utf-8-sig，这样才能带BOM头。如果你用UTF-8不带BOM，Excel默认用本地字符集解析，就会乱。所以：

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

如果文件特别大，比如几个GB，不要用Pandas一次性读入内存再转存。正确做法是分块读取、分块处理、分块写出。比如从数据库里按主键分页拉取，每次取5万行，清洗完追加到目标文件。这样内存占用始终可控，程序也不会因为内存爆炸而挂掉。

还有一个容易被忽略的问题：重复跑同一个任务的时候，怎么保证没有重复数据？我的建议是“先查后插”。在写入目标表之前，先按主键字段查询目标表已有的ID集合，然后只插入不存在的记录。如果主键不是单一字段，那就拼接一个业务主键字符串来做对比。这个方法很简单，但能保证数据的幂等性，比单纯依赖drop_duplicates稳妥得多。

5. 高频问题与排查技巧实录

数据处理与获取过程中，真正让人头秃的往往不是哪种算法高级，而是那些“莫名其妙”的问题。我挑几个频率最高的，把排查思路和对应的处理方式都列出来。

5.1 请求类问题的判断顺序

遇到403：先确认是不是没有带User-Agent或Referer，然后检查是否需要登录后的Cookie。在自己账号权限允许的前提下，可以登录后把Cookie放进请求头，但还是那句话，一切以数据源方的条款和授权为准。另外，有些服务器的403是安全策略拦截了不常见的行为特征，比如请求头顺序、HTTP版本、指纹一致性问题。先把请求头里的多余字段删掉，只保留最基础的几个，往往就能解决。

遇到429：说明请求太频繁，被限流了。我试过一边sleep 1秒一边请求还是被限，后来发现是并发线程太多。解决办法是把请求间隔拉大，并且把任务分批，每批跑完休息一分钟。没有万能参数，核心是在不影响自己和别人正常使用的前提下，做一个足够有礼貌的调用者。

页面内容为空：先确认返回的text里有没有你要的关键词。如果HTML里没有，大概率数据是动态加载的。这时回到第1部分讲的，去网络面板里找XHR/JSON接口。如果接口也找不到，看看是不是内容在<script>标签里以window.__DATA__这样的全局变量形式存在，可以用正则提取。

5.2 数据清洗类问题的经典场景

全表日期解析失败：多半是日期格式里混杂了null、空字符串或者非标准写法。可以先用正则把原始字符串里的日期片段提取出来，再交给to_datetime。比如先把所有分隔符统一成-，把中文年月日替换掉，再解析，成功率会高很多。

数值列突然出现大量NaN：除了原始数据确实为空，更常见的是类型转换失败。比如一个“金额”列里既有数字又有“待定”这样的字符，pd.to_numeric就会把整列转不过来。我的做法是保留原始列，新建一个标准化列，并且统计转换失败的行数，避免查到一半才发现大量数据被置空。

处理完数据行数对不上：要么是去重过于激进，要么是解析时漏掉了某页。建议清洗前后都打印一个shape，并且把主键的次数分布打出来，看看哪行数据被重复计数或丢失。

5.3 程序稳定性问题

程序跑一半断了：这种事情不可避免。我的解决方案是“断点续跑”：把已经成功获取并解析的数据实时落盘成CSV/JSON文件，每次启动任务时先判断本地是否已经存在结果文件，如果存在，就跳过那些已经处理过的记录。这个思路不仅防止重复请求，也能让你在程序意外挂掉时不会一夜回到解放前。

内存占用过高：首先看是因为数据量太大还是代码里放了大量的中间变量。排查方法是把循环体里创建的临时变量清掉，或者干脆用生成器逐行迭代，而不是把所有结果都append到一个大列表里再转DataFrame。实在要用列表暂存，那就每处理一万行就把这个列表转存成文件并清空列表。

字段顺序乱了：多文件合并时，两个文件可能字段顺序不一样。用pandas合并时别依赖位置，要用列名对齐，比如pd.concat([df1, df2], join="outer")，然后手动调整最终列顺序。

我整理了一张速查表，方便大家遇到问题的时候快速对照：

6. 把整套流程变得可维护：日志、调度与自检

写脚本跑一次，跟写一套可持续运行的流程，差别很大。前者随时能跑通一次就够了，后者却要求你有“长期主义”的思维，哪怕数据源结构变了、服务器抽风了、某个字段没了，你也能第一时间发现并定位。

6.1 日志与运行状态记录：别靠肉眼盯控制台

现在的数据处理脚本多少会有一点定时执行的需求，人不可能天天守在电脑前看它跑完。所以我习惯在每个阶段都输出结构化的日志，至少包含时间、任务名、状态、影响行数。不要用一堆print，用logging模块就够了，再把日志写到文件里：

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("etl.log", encoding="utf-8"), logging.StreamHandler() ] ) logging.info("开始拉取第1页数据，当前记录数：%d", start_count)

等到脚本跑完，再看一眼“成功数”和“失败数”。如果失败率突然变高，通常不是网络波动，而是页面结构变了。

6.2 定时执行与增量更新：用最简单的方式养成习惯

我见过很多团队硬上Airflow等重量级调度框架来处理日常几万行的数据任务，其实有点杀鸡用牛刀。对绝大多数个人或小团队场景，直接用系统自带的cron按时跑脚本就够了：

# 每天凌晨2点执行一次 0 2 * * * cd /data/project && python main.py >> logs/cron.log 2>&1

如果要用在执行机上跑并且需要任务编排，还是要保证脚本是“可重复执行”的。具体做法是：每次执行前，先记录开始时间，执行期间处理逻辑只处理某个时间窗口之后的新数据，执行完后把时间窗口往前推一格，存成状态文件。这样就算某天任务重跑，也不会重复处理历史数据。

增量更新的核心思想从来不是写多复杂的框架，而是“主键+时间戳”。只要目标表里存在可对比的主键，就用“存在则更新，不存在则插入”的方式。用SQL表达就是先查一遍已有主键，把剩下的新增数据批量插入。

6.3 数据正确性自检：跑完不代表跑对

长期运行的数据任务最怕“每次都能跑成功但数据结果悄悄变了”。所以我每次跑完之后，都会加一个自检环节，一般包含三个动作：

数量级核对。比如今天的记录数跟昨天比，浮动超过30%就要告警。就算业务真的有波动，也要人工确认一遍。

关键字段抽样。随机抽取十行数据，打印原始页面里的样本与清洗后表格里的样本，人工或脚本比对几个核心字段是否一致。

schema校验。检查输出DataFrame的列名是否跟预期的标准schema完全一致，多一列少一列都要输出来。这一步可以用简单的断言实现：

assert set(required_cols).issubset(df.columns), f"缺列: {required_cols - set(df.columns)}"

这一套下来，数据正确性有底，后面别人问起来也能拿日志和自检记录说明结果可信。

在这几年的实操经历里，我自己最大的体会是：做数据处理工作，千万不要在一开始追求“完美代码”，而是要把“可复现、可排查、可恢复”放在第一位。数据源是变化的，业务需求也是变化的，一套流程里最值钱的并不是那几行爬虫代码，而是你在运行过程中沉淀下来的日志、状态文件、清洗规则和排错经验。每次处理一份新数据，我都会先把原始数据原封不动存一份，再另存一版清洗后的数据。两个版本都留着，后面哪怕发现清洗规则写错了，也能从原始数据重新生成一次，而不会因为源头数据被覆盖而无从下手。

最后再分享一个小技巧：所有清洗规则尽量写成配置文件或者独立的函数，不要散落在各种脚本里。把“金额单位统一” “日期格式标准”这类逻辑抽出来，下次遇到同类型数据还能复用。数据获取和处理这件事，做得多了你会发现，稳定比速度重要，规则清晰比代码炫技重要。希望对正在跟数据打交道的你有帮助。