1. 从一份嵌套 JSON 说起Python 写 MySQL 到底难在哪如果你手上有一份爬下来的 JSON 文件想把它塞进 MySQL大概率会经历这么几个阶段先json.loads解析发现字段对不上表结构改完字段又发现嵌套的那层actors是个数组直接塞进去报类型错误好不容易跑通了单条插入几万行数据跑得像蜗牛。这篇就围绕「Python 将 JSON 数据存入 MySQL」这条链路把解析、字段映射、批量入库、事务提交一次讲透最后用一条带嵌套结构的样例数据做端到端验证确认行数和字段值都对得上。先说清楚适用人群会一点 Python 基础语法、装过 MySQL、能看懂SELECT和INSERT就够。不需要你会 ORM也不需要你懂连接池全程用pymysql手写 SQL因为这样最容易看清每一步发生了什么。JSON 是什么、能做什么这里不展开你既然搜到这篇说明你已经有数据在手缺的是「怎么落库」这一步。我先把整体链路摆出来后面每一节都对应其中一环JSON 文件 → json.loads 解析 → 字段映射/扁平化 → 建表 SQL → pymysql 连接 → executemany 批量插入 → commit 提交 → 验证行数这条链路里最容易翻车的是两处一是嵌套结构没拍平就直接往varchar字段塞二是executemany的参数写成了一维列表而不是「元组的列表」。下面逐个拆。先给一条本文贯穿使用的样例数据假设文件叫movies.json每行一个 JSON 对象JSONL 格式爬虫导出常见形态{name: 示例电影A, releasetime: 2023-05-01, score: 8.7, actors: [演员甲, 演员乙, 演员丙]} {name: 示例电影B, releasetime: 2022-11-15, score: 7.9, actors: [演员丁, 演员戊]}注意actors是数组这就是嵌套结构。你可以选择把它json.dumps成字符串存进一个字段也可以拆成关联表。本文为了让你先跑通主链路采用「数组转字符串存 TEXT 字段」的方案简单、可验证后面再提拆表思路。建表 SQL 先给出来字段类型和样例数据严格对应CREATE TABLE IF NOT EXISTS movies ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) NOT NULL, releasetime VARCHAR(32) NOT NULL, score VARCHAR(16) NOT NULL, actors TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;utf8mb4这个字符集别省中文和特殊符号都靠它。score用VARCHAR而不是DECIMAL是因为爬下来的分数经常带「暂无评分」这种脏值先存字符串清洗逻辑放到入库前做避免插入时直接报错中断。到这里问题场景和数据结构都清楚了。接下来解决「用什么连、怎么连」的问题。2. TaoToken 前置准备把模型能力和数据库链路接起来写这类数据管道脚本很多时候不是一次就能写对的。字段名拼错、SQL 占位符数量对不上、嵌套解析抛异常这些都需要你一边写一边调。如果你习惯让模型帮你补全executemany的写法、或者解释某条报错那在动手前把调用入口配好会省不少来回。TaoToken 在这里的角色是统一的模型调用入口官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。它不替代你的 MySQL也不替代pymysql它替代的是你「遇到不会写的 SQL 或解析逻辑时去翻文档」的那段时间。你可以把它理解成一个随时能问的结对伙伴。具体怎么用取决于你的工作方式。如果你是在编辑器里写脚本想让模型直接读你的代码上下文可以走 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。如果你只是想临时问一句「pymysql 的 executemany 参数格式是什么」用模型对话就够了入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。这里要强调一个配置三件套的概念不管你用哪种客户端接入任何模型服务都绕不开这三个值Base URL、API Key、Model ID。以常见的 OpenAI 兼容客户端为例配置长这样{ base_url: https://taotoken.net/api, api_key: sk-你的密钥, model: claude-sonnet-4-5 }base_url填 https://taotoken.net/api 注意末尾不要多加/v1之类的路径具体以接入文档为准。api_key在控制台生成入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。model填你要用的模型 ID不同客户端字段名可能叫model或model_id以文档为准文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果你用的是 Claude Code 这类命令行工具配置方式又不一样它走的是环境变量或配置文件。对应的接入说明在 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 里面会告诉你ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY怎么填。控制台入口统一在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 密钥管理、用量查看都在那里。为什么要在数据库教程里花篇幅讲这个因为实际写管道时你遇到的报错往往不是「不会写」而是「写错了但看不出哪错」。比如pymysql.err.ProgrammingError: (1064, ...)这种模型能帮你快速定位是占位符数量不匹配还是字段名拼写问题。把入口配好调试效率会明显不一样。前置准备就这些不涉及任何数据库侧的改动。下面进入正题开始写可复制的配置和脚本。3. 可复制配置连接参数、建表 SQL 与批量插入脚本这一节是全文的核心所有代码都可以直接复制改路径使用。我按「连接配置 → 建表 → 解析 → 批量插入 → 提交」的顺序给每一步都说明为什么这么写。先看连接配置。pymysql.connect的参数建议单独抽成字典方便你改import pymysql import json DB_CONFIG { host: localhost, port: 3306, user: root, password: 你的密码, database: json_demo, charset: utf8mb4, autocommit: False, }autocommit设成False是关键。批量插入时我们希望所有行要么全成功、要么全回滚手动commit才能保证这一点。如果你设成True每插一行就提交一次中途报错会留下半截数据验证行数时对不上排查起来很痛苦。建表函数def init_table(conn): with conn.cursor() as cursor: cursor.execute( CREATE TABLE IF NOT EXISTS movies ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) NOT NULL, releasetime VARCHAR(32) NOT NULL, score VARCHAR(16) NOT NULL, actors TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) conn.commit()解析与字段映射函数。这里处理嵌套的actors数组用json.dumps转成字符串def parse_line(line): obj json.loads(line) return ( obj[name], obj[releasetime], obj[score], json.dumps(obj.get(actors, []), ensure_asciiFalse), )ensure_asciiFalse保证中文不被转成\uXXXX存进数据库后直接可读。obj.get(actors, [])用get而不是[]是为了防止某行缺字段直接抛KeyError中断整个流程。批量插入函数这是最容易写错的地方def batch_insert(conn, rows, batch_size1000): sql INSERT INTO movies (name, releasetime, score, actors) VALUES (%s, %s, %s, %s) total 0 with conn.cursor() as cursor: for i in range(0, len(rows), batch_size): chunk rows[i:i batch_size] try: cursor.executemany(sql, chunk) conn.commit() total len(chunk) print(f已提交 {total} 行) except Exception as e: conn.rollback() print(f第 {i} 行起插入失败: {e}) raise return total注意executemany的第二个参数必须是「元组的列表」也就是[(v1,v2,v3,v4), (v1,v2,v3,v4), ...]。如果你传成[v1, v2, v3, v4]这种一维列表会报ProgrammingError。这是新手最常见的坑我在第 5 节会专门对照报错讲。主流程串起来def main(): rows [] with open(movies.json, encodingutf-8) as f: for line in f: line line.strip() if not line: continue rows.append(parse_line(line)) conn pymysql.connect(**DB_CONFIG) try: init_table(conn) inserted batch_insert(conn, rows) print(f共写入 {inserted} 行) finally: conn.close() if __name__ __main__: main()分块batch_size1000是个经验值。太小提交频繁太大单次事务日志膨胀1000 到 5000 之间都算合理。数据量上万时分块提交还能避免一个超长事务把连接拖死。如果你想让模型帮你检查这段脚本的字段映射是否和建表 SQL 一致可以把两段代码一起贴给它让它逐字段比对。这种「代码审查」式的提问比单独问「executemany 怎么用」效率高得多。配置和脚本都给全了下面跑一遍看结果。4. 端到端验证跑通请求、确认行数与字段值一致验证分两步先确认脚本能跑通再确认数据库里的数据对得上。别跳过第二步很多人脚本没报错就以为成了结果字段错位、中文乱码等到用数据时才发现。先准备movies.json内容用第 1 节给的两行样例。然后执行python insert_json.py预期输出已提交 2 行 共写入 2 行如果看到这个说明插入链路通了。接下来进 MySQL 验证。用命令行或者你习惯的客户端都行SELECT id, name, releasetime, score, actors FROM movies;预期结果idnamereleasetimescoreactors1示例电影A2023-05-018.7[演员甲, 演员乙, 演员丙]2示例电影B2022-11-157.9[演员丁, 演员戊]行数是 2和 JSON 文件里的行数一致。字段值逐个核对name是中文没乱码releasetime格式没被改动score保持字符串actors是 JSON 数组序列化后的字符串。这几项都对说明解析、映射、插入三步都没问题。再补一条聚合验证确认没有重复插入SELECT COUNT(*) AS total, COUNT(DISTINCT name) AS uniq_name FROM movies;如果total和uniq_name相等说明没有因为重跑脚本产生重复数据。这里要提醒一句本文的建表用了CREATE TABLE IF NOT EXISTS重跑脚本不会清表会追加数据。如果你需要幂等把建表改成先DROP TABLE IF EXISTS再CREATE或者给name加唯一索引配合INSERT ... ON DUPLICATE KEY UPDATE。这个取舍看你的场景爬虫增量入库通常用后者。验证通过后你可以把actors字段取出来反序列化确认能还原成数组import json # 假设从数据库取出的 actors 字段值为 s s [演员甲, 演员乙, 演员丙] print(json.loads(s)) # [演员甲, 演员乙, 演员丙]能还原说明序列化时用的ensure_asciiFalse生效了中文没被转义。这一步很多人会漏等到前端展示时才发现存进去的是\u6f14\u5458\u7532再回头改就麻烦了。端到端验证到这里就完整了。下面把过程中可能遇到的报错集中列一下。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth这一节的报错分两类一类是数据库侧的一类是模型调用侧的。两类都会在你写管道时遇到分开说。数据库侧最常见的三个报错一pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax...)九成是executemany的参数格式错了。检查你的rows是不是「元组的列表」。错误写法rows [电影A, 2023-05-01, 8.7, []] # 一维列表错 cursor.executemany(sql, rows)正确写法rows [(电影A, 2023-05-01, 8.7, [])] # 元组的列表对 cursor.executemany(sql, rows)另一个可能是占位符数量对不上。SQL 里 4 个%s元组里就必须 4 个值多一个少一个都报 1064。报错二pymysql.err.OperationalError: (1045, Access denied for user...)用户名或密码错或者该用户没有目标库的权限。先确认DB_CONFIG里的user和password再确认database存在。如果库不存在先CREATE DATABASE json_demo DEFAULT CHARSET utf8mb4;。报错三中文乱码存进去是问号检查三处字符集是否统一连接参数charsetutf8mb4、建表DEFAULT CHARSETutf8mb4、读取文件encodingutf-8。三处任一不一致都可能乱码。另外json.dumps记得加ensure_asciiFalse。模型调用侧的报错如果你按第 2 节配了 TaoToken可能会碰到这几个报错四401 UnauthorizedAPI Key 无效或没带上。检查请求头里的Authorization: Bearer sk-xxx是否完整Key 是否在控制台被禁用或删除。重新生成一个 Key 再试入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。报错五local proxy failed或连接超时这类通常是 Base URL 填错或者本地网络环境有额外配置。确认base_url是 https://taotoken.net/api 不要自己拼/v1/chat/completions之类的完整路径客户端一般会自动补。如果还是不通对照接入文档检查文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。报错六reading choices相关解析失败这通常出现在流式响应处理里客户端期望的响应结构和实际返回不一致。先确认你用的 Model ID 是否被该客户端支持再确认是否开了流式。非流式请求先跑通再加流式。Model ID 的可用列表以文档为准。报错七OAuth 相关报错如果你用的是 Claude Code 这类走 OAuth 的工具报错里出现OAuth字样通常是认证流程没走完或 token 过期。按 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 里的步骤重新配置环境变量注意ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY两个都要填缺一个都会失败。排查这类问题的通用思路是先确认三件套Base URL、Key、Model ID齐全且正确再看客户端日志里的完整请求 URL 和响应体。把这两样贴给模型定位速度会快很多。数据库和模型两侧的坑都列完了最后说下这套链路怎么长期用。6. 把这条链路用起来从单文件脚本到稳定管道跑通一次不难难的是让它稳定地跑下去。这里给几个我实际用下来有效的做法。第一把解析和入库解耦。解析函数只负责「JSON 行 → 元组」入库函数只负责「元组列表 → 数据库」中间用列表传递。这样你换数据源时只改解析换数据库时只改入库互不影响。本文的parse_line和batch_insert就是按这个思路拆的。第二加日志而不是只加 print。数据量上来后你需要知道「哪一批失败了、失败在第几行」。把print换成logging记录批次起始行号和异常堆栈出问题时能直接定位到具体数据。第三脏数据前置过滤。爬下来的 JSON 经常有缺字段、类型不对的行。在parse_line里加校验缺关键字段的行直接跳过并记录不要让它进到executemany里。一条脏数据能让整批回滚代价太大。第四大批量用分块提交。本文的batch_size1000是起点你可以根据单行大小调整。分块的好处是失败时只回滚当前块前面的数据已经落库重跑时从断点继续即可。第五验证脚本常态化。每次入库后跑一遍COUNT(*)和字段抽样确认行数和内容都对。这一步花不了几秒但能帮你早发现字段错位、编码问题。如果你在写这套管道时需要模型帮你补全某个函数、解释某条报错或者审查字段映射可以按第 2 节的入口配置。长期做数据工程的话Coding Plan 会比单次对话更顺手入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。临时验证某个模型对 JSON 解析的理解用模型对话就够入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。最后留一个可以立刻动手的练习把本文的movies.json换成你手上真实的爬虫数据先跑 10 行确认字段映射对得上再放开全量。遇到字段对不上时别急着改 SQL先把解析后的元组打印出来看一眼多数问题在那一行就能看出来。
