Python 爬虫数据处理：多表关联整合多爬虫来源异构业务数据集

前言

多源异构数据采集是规模化爬虫项目落地后的核心收尾环节，同一业务指标往往经由资讯爬虫、商品爬虫、评论爬虫、接口抓取爬虫、小程序抓包爬虫多条链路分散采集，原始数据分储于 CSV、Excel、MySQL 单表、SQLite、JSON 零散文件等多种存储载体，字段命名规则、数据单位、时间格式、编码规范、空值填充逻辑互不统一，直接入库会出现主键冲突、维度断层、统计口径错乱、冗余字段泛滥等问题。多表关联整合依托数据清洗标准化、主键映射绑定、多维度关联算法、分库分表数据拉取机制完成异构数据归一化，打通多爬虫数据源的数据壁垒，形成结构化宽表数据集，为后续数据分析、报表生成、BI 可视化、指标建模提供合规可用的标准数据源。本文围绕异构数据预处理规范、三种主流关联匹配算法、多存储介质联合读取实操、批量数据合并入库全流程展开，搭配实测对照表格与分段可运行 Python 源码，覆盖文件型异构、数据库跨库异构、文件 + 数据库混合异构三类高频业务整合场景。

前置依赖库资源超链接汇总

结构化数据处理 Pandas 数值高效计算 Numpy MySQL 数据库连接 PyMySQL SQLite 内置标准库 sqlite3 Excel 读写 openpyxl JSON 数据解析标准库 json 数据库连接通用工具 SQLAlchemy 大数据分块读取 Dask

一、多爬虫异构数据源特征与数据整合痛点梳理

1.1 多爬虫数据源异构差异化明细

多爬虫项目开发过程中，不同爬虫由不同开发周期、不同开发人员迭代开发，采集规则随站点改版持续迭代，最终落地的原始数据在存储格式、字段体系、数据内容上形成天然异构，下表罗列四类主流爬虫产出数据源的核心异构特征。

表格

1.2 数据整合四大核心痛点

第一，关联主键不统一：同一商品在商品爬虫中为 spu_id、评论爬虫为 goods_id、活动爬虫为 product_no，编码规则位数、前缀字符不一致，无法直接通过等值条件关联；第二，脏数据格式紊乱：同一价格字段出现59、59.0元、¥59、59RMB多种书写格式，无法做数值运算与聚合；第三，存储介质割裂：数据分散在文件与不同数据库实例，常规 SQL 联表查询无法跨文件、跨数据库读取数据；第四，数据增量冲突：多爬虫异步采集，同一条数据多次更新，出现重复主键、新旧数据并存，合并后产生冗余脏数据。

想要实现数据规整合并，整体流程划分为：多源数据统一读取→全字段标准化清洗→主键别名映射归一→多条件关联拼接→重复数据去重→标准化宽表落地存储六大环节。

二、异构数据预处理标准化规范与代码落地

预处理是多表关联前置基础，未标准化的数据直接关联会造成大量匹配失效、关联空值占比超标，预处理分为字段格式清洗、空值统一处理、主键标准化改写三个模块。

2.1 通用数据清洗工具类封装

搭建统一清洗工具类，集成价格去符号、日期格式统一、空白字符剔除、空值归一四类通用清洗方法，后续所有数据源统一调用工具函数完成字段规整。

python

运行

import re import pandas as pd from datetime import datetime class DataCleanTool: @staticmethod def clean_price(raw_price): """价格字段清洗：剔除人民币符号、中文，保留浮点数字""" if pd.isna(raw_price) or str(raw_price).strip() in ["", "-", "无", "null"]: return None raw_str = re.sub(r"[¥元RMB￥]", "", str(raw_price)) num_match = re.search(r"\d+\.?\d*", raw_str) if num_match: return float(num_match.group()) return None @staticmethod def clean_date(raw_date): """统一转换为%Y-%m-%d标准日期格式，无法转换返回空""" if pd.isna(raw_date) or str(raw_date).strip() in ["", "-", "无"]: return None raw = str(raw_date).strip() fmt_list = ["%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"] for fmt in fmt_list: try: return datetime.strptime(raw, fmt).strftime("%Y-%m-%d") except ValueError: continue # 时间戳转换逻辑 if raw.isdigit() and len(raw) in [10,13]: ts = int(raw[:10]) return datetime.fromtimestamp(ts).strftime("%Y-%m-%d") return None @staticmethod def clean_blank(raw_text): """去除首尾空格、全角空格，空字符统一转为pandas空值""" if pd.isna(raw_text): return None res = re.sub(r"\s+", "", str(raw_text)) return res if res else None # 工具实例化 clean_tool = DataCleanTool()

代码底层原理说明

1. clean_price 依托正则表达式匹配字符串内有效数字，剔除各类货币符号与中文单位，统一输出浮点型数值，解决价格字段多格式异构问题；
2. clean_date 预设市面上爬虫最常见的三类日期字符串格式，依次尝试格式化转换，额外兼容 10 位 / 13 位 Unix 时间戳，最终全部统一为YYYY-MM-DD字符串标准格式；
3. clean_blank 使用正则批量替换各类空白占位符，将空字符串、全空格数据统一转为 pandas 空值，实现全库空值口径统一。

2.2 主键标准化映射处理

不同爬虫主键命名不一致是关联失败的首要因素，采用映射字典建立别名对照表，将 spu_id、goods_id、product_no 等不同字段名统一映射为标准主键std_spu，完成主键归一。

python

运行

# 主键映射配置字典，key为原始字段名，value为标准字段名 primary_map = { "spu_id": "std_spu", "goods_id": "std_spu", "product_no": "std_spu", "shop_code": "std_shop", "store_id": "std_shop" } def rename_primary_col(df:pd.DataFrame, col_map:dict): """批量重命名主键字段，非主键字段保持原有名称""" rename_dict = {} for old_name, new_name in col_map.items(): if old_name in df.columns: rename_dict[old_name] = new_name df = df.rename(columns=rename_dict) return df

代码底层原理说明

函数遍历映射字典，仅对当前数据集已存在的字段执行重命名，避免因字段缺失导致的报错，全部数据源经过该函数处理后，商品统一主键为 std_spu、店铺统一主键 std_shop，后续多表可基于统一主键做关联运算。

三、多数据源读取实现：文件 + 数据库混合异构数据载入

分别实现 MySQL 数据表读取、CSV 批量分片读取、Excel 单文件读取、本地 JSON 文件夹批量读取，将四类异构数据全部转为 Pandas DataFrame 内存结构化表格，完成多源数据同格式落地，为关联拼接做准备。

3.1 MySQL 商品基础表读取代码

python

运行

import pymysql from sqlalchemy import create_engine def read_mysql_spu(table_name:str, db_conf:dict): """读取MySQL商品基础数据表""" conn_str = f"mysql+pymysql://{db_conf['user']}:{db_conf['pwd']}@{db_conf['host']}:{db_conf['port']}/{db_conf['db']}" engine = create_engine(conn_str) sql = f"SELECT spu_id,price,origin,putaway_date FROM {table_name}" df_mysql = pd.read_sql(sql, engine) # 字段清洗 df_mysql["price"] = df_mysql["price"].apply(clean_tool.clean_price) df_mysql["putaway_date"] = df_mysql["putaway_date"].apply(clean_tool.clean_date) # 主键标准化重命名 df_mysql = rename_primary_col(df_mysql, primary_map) return df_mysql # 数据库连接配置 mysql_config = { "host":"127.0.0.1", "user":"root", "pwd":"root123", "port":3306, "db":"spider_goods" } df_goods = read_mysql_spu("goods_info", mysql_config)

3.2 批量 CSV 评论分片文件读取

评论爬虫按日期拆分数十个 CSV 文件，遍历指定文件夹批量读取并合并为单张 DataFrame：

python

运行

import os def read_batch_csv(csv_dir:str): """遍历目录读取全部csv评论文件，合并数据集""" csv_list = [os.path.join(csv_dir,file) for file in os.listdir(csv_dir) if file.endswith(".csv")] df_list = [] for path in csv_list: temp_df = pd.read_csv(path, encoding="utf-8", usecols=["goods_id","star","comment_time","content"]) df_list.append(temp_df) df_comment = pd.concat(df_list, ignore_index=True) # 字段清洗 df_comment["comment_time"] = df_comment["comment_time"].apply(clean_tool.clean_date) df_comment["content"] = df_comment["content"].apply(clean_tool.clean_blank) df_comment = rename_primary_col(df_comment, primary_map) return df_comment df_comment = read_batch_csv("./comment_csv/")

3.3 Excel 店铺台账数据读取

python

运行

def read_excel_shop(excel_path:str): df_shop = pd.read_excel(excel_path, engine="openpyxl", usecols=["shop_code","address","license_no"]) df_shop["address"] = df_shop["address"].apply(clean_tool.clean_blank) df_shop = rename_primary_col(df_shop, primary_map) return df_shop df_shop = read_excel_shop("./shop_info.xlsx")

3.4 批量 JSON 活动爬虫数据读取

活动爬虫 JSON 为嵌套字典格式，需要平铺嵌套字段后再标准化：

python

运行

import json def read_batch_json(json_dir:str): json_files = [os.path.join(json_dir,f) for f in os.listdir(json_dir) if f.endswith(".json")] data_list = [] for path in json_files: with open(path,"r",encoding="utf-8") as f: raw_data = json.load(f) # 平铺嵌套商品ID数组 for item in raw_data: pro_list = item.get("rel_product",[]) for pid in pro_list: new_row = { "product_no":pid, "discount":item.get("discount"), "full_cut":item.get("full_money") } data_list.append(new_row) df_act = pd.DataFrame(data_list) df_act = rename_primary_col(df_act, primary_map) return df_act df_activity = read_batch_json("./act_json/")

整体读取逻辑原理说明

所有数据源经过读取→字段清洗→主键标准化三步处理后，商品、评论、店铺、活动四张数据集统一以 std_spu 作为商品关联主键，std_shop 作为店铺关联主键，存储格式统一为 DataFrame，彻底抹平原始存储载体带来的介质壁垒。

四、三大多表关联算法详解与实战落地

Pandas 内置 merge、join、concat 三类关联方式，分别对应 SQL 多表内联、外联、纵向追加场景，结合爬虫业务特征区分使用场景，表格对比三种关联方式的适用范围。

表格

4.1 Merge 多主键左关联（业务最常用，商品 + 评论 + 活动三表联查）

爬虫业务优先选用左连接，以商品主表为基准，左联评论、活动数据，保留全量商品基础数据，无匹配评论 / 活动的商品对应字段自动填充空值，和 MySQL LEFT JOIN 逻辑完全一致。

python

运行

# 第一步：商品左联评论 df_merge1 = pd.merge( left=df_goods, right=df_comment, on="std_spu", how="left" ) # 第二步：合并结果继续左联活动数据表 df_final = pd.merge( left=df_merge1, right=df_activity, on="std_spu", how="left" )

代码原理说明

how="left" 代表左连接，左表所有数据全部保留，右表仅匹配主键一致的数据并入；on 指定关联字段为统一后的标准主键 std_spu，经过两次 merge 完成商品、评论、促销数据三合一宽表。

4.2 Join 索引关联（店铺数据快速挂载）

将店铺数据表以 std_shop 设为索引，使用 join 快速挂载至主数据集，适用于维度字典类小表关联：

python

运行

df_shop = df_shop.set_index("std_shop") df_final = df_final.join(df_shop, on="std_shop", how="left")

4.3 Concat 纵向拼接（增量爬虫新增数据合并）

每日增量爬虫产出新 CSV 评论文件，字段结构和历史存量数据完全一致时，使用 concat 纵向追加新增数据：

python

运行

# 新采集增量评论数据 df_new_comment = read_batch_csv("./new_comment/") # 历史存量评论 df_old_comment = df_comment # 纵向合并 df_all_comment = pd.concat([df_old_comment, df_new_comment], ignore_index=True)

五、关联后数据去重与异常数据剔除

多爬虫异步采集极易产生重复数据，分为主键全重复、部分字段重复两类，分别使用 drop_duplicates、分组聚合两种方案处理。

5.1 全字段重复直接去重

python

运行

# 全字段完全一致直接删除重复行，保留首次出现数据 df_final = df_final.drop_duplicates(keep="first")

5.2 同一商品多条评论 / 多条活动数据分组聚合

一条商品对应多条评论、多条促销活动，基于 std_spu 分组，评论内容、折扣字段以列表聚合存储，避免主键多行拆分造成主表数据膨胀：

python

运行

def agg_func(x): res_dict = {} res_dict["comment_content"] = list(x["content"].dropna()) res_dict["star_avg"] = round(x["star"].mean(),2) if not x["star"].isna().all() else None res_dict["discount_list"] = list(x["discount"].dropna()) return pd.Series(res_dict) # 分组聚合，合并多条明细数据 agg_col = ["content","star","discount"] group_df = df_final.groupby("std_spu")[agg_col].apply(agg_func).reset_index() # 聚合结果回写主表 final_wide = pd.merge(df_goods.drop(agg_col,axis=1), group_df, on="std_spu", how="left")

聚合逻辑原理说明

以标准商品主键分组，同商品多条明细数据转为列表格式存入单行，既保留全量明细信息，又保证最终宽表一行对应一个商品，适配后续入库与统计需求。

六、标准化宽表多方式落地存储（入库 MySQL / 归档 CSV）

整合完成的标准宽表支持两种落地方案，一是写入业务 MySQL 正式数据表用于业务查询，二是导出 CSV 归档备份用于离线数据分析。

6.1 批量写入 MySQL 正式业务表

python

运行

def save_to_mysql(df:pd.DataFrame, table:str, conf:dict): conn_str = f"mysql+pymysql://{conf['user']}:{conf['pwd']}@{conf['host']}:{conf['port']}/{conf['db']}" engine = create_engine(conn_str) df.to_sql(name=table, con=engine, index=False, if_exists="replace", chunksize=2000) save_to_mysql(final_wide, "spider_union_goods", mysql_config)

参数说明

if_exists="replace" 表示全量覆盖写入，增量更新场景可改为 append 追加写入，chunksize 控制单次入库行数，避免大数据量瞬间占用数据库 IO 资源。

6.2 本地归档导出 CSV

python

运行

final_wide.to_csv("./union_all_data.csv", index=False, encoding="utf-8-sig")

七、超大体量异构数据优化方案（Dask 分块处理）

单表数据突破百万行后，Pandas 全量载入内存会出现内存溢出，使用 Dask DataFrame 分块懒加载，分批次完成清洗与关联运算，无需一次性加载全量数据至内存。

python

运行

import dask.dataframe as dd # Dask分块读取海量CSV评论 dask_comment = dd.read_csv("./comment_csv/*.csv", usecols=["goods_id","star","comment_time","content"], blocksize="10MB") # 批量清洗 dask_comment["comment_time"] = dask_comment["comment_time"].apply(clean_tool.clean_date, meta=("comment_time","object")) # 主键重命名 dask_comment = dask_comment.rename(columns={"goods_id":"std_spu"}) # 执行运算落地pandas df_big_comment = dask_comment.compute()

Dask 底层原理说明

Dask 采用分块存储 + 延迟计算机制，读取数据时按照设置的 blocksize 切分数据块，所有清洗、关联逻辑仅生成任务图，调用 compute () 后才逐块执行运算，显著降低峰值内存占用，适配千万级爬虫原始数据整合场景。

八、实测整合效果对照数据表

选取 12000 条商品数据、56000 条评论数据、3200 条活动数据开展整合测试，对比标准化前后数据匹配率、空值占比关键指标：

表格

从实测数据可见，完整的预处理 + 主键映射流程可将多源数据有效可用率由不足 30% 提升至 93% 以上，是爬虫数据落地业务库不可或缺的标准化步骤。

九、项目落地高频问题与优化对策

9.1 问题 1：部分主键编码格式混杂字符串前缀（S001/001）导致匹配失败

优化方案：新增主键清洗函数，通过正则剔除主键非数字前缀，统一编码格式后再做关联。

python

运行

def clean_spu_code(code): if pd.isna(code): return None num = re.search(r"\d+",str(code)) return num.group() if num else None

9.2 问题 2：数据源存在百万级超大表，merge 关联内存溢出

优化方案：大表按日期分块拆分，循环分块与小表关联，分次拼接结果；或切换 Dask 分块关联。

9.3 问题 3：增量每日新增爬虫数据重复入库

优化方案：入库前以 std_spu 作为唯一主键，先根据主键删除存量数据，再追加当日新数据，实现幂等入库。

结语

多爬虫异构数据整合是连接原始爬虫采集与上层数据应用的关键枢纽，核心思路围绕统一字段格式、统一关联主键、选用适配关联算法三大要点落地。本文实现从多格式原始数据读取、全链路清洗标准化、多条件关联拼接、去重聚合到落地入库的全流程可复用代码，整套工具类可直接嵌入现有爬虫项目，适配从万级到千万级不同体量爬虫数据整合需求。在后续项目迭代中，可结合 SQLAlchemy 配置多数据源连接池、借助 Airflow 搭建定时自动整合调度任务，实现爬虫数据落地全自动化运行，进一步降低人工整理成本。