2025-06-09 约 2400 字 预计阅读 5 分钟
数据处理问题
数据采集、解析和存储中的常见问题及解决方案。
编码错误
UnicodeDecodeError
问题表现 :
1
UnicodeDecodeError : 'utf-8' codec can 't decode byte 0xff in position 0
原因 : 文件或数据使用了不同的编码格式
解决方案 :
1. 自动检测编码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import chardet
# 检测文件编码
with open ( 'file.txt' , 'rb' ) as f :
raw_data = f . read ()
result = chardet . detect ( raw_data )
encoding = result [ 'encoding' ]
confidence = result [ 'confidence' ]
print ( f "Encoding: { encoding } (Confidence: { confidence } )" )
# 使用检测到的编码读取
with open ( 'file.txt' , 'r' , encoding = encoding ) as f :
content = f . read ()
2. 处理网页编码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import requests
from chardet import detect
response = requests . get ( url )
# 方法 1: 让 requests 自动检测
response . encoding = response . apparent_encoding
text = response . text
# 方法 2: 手动检测
encoding = detect ( response . content )[ 'encoding' ]
text = response . content . decode ( encoding , errors = 'ignore' )
# 方法 3: 尝试常见编码
encodings = [ 'utf-8' , 'gbk' , 'gb2312' , 'gb18030' , 'big5' ]
for enc in encodings :
try :
text = response . content . decode ( enc )
break
except UnicodeDecodeError :
continue
3. 忽略错误字符
1
2
3
4
5
6
7
8
# 忽略无法解码的字符
text = data . decode ( 'utf-8' , errors = 'ignore' )
# 或替换为 ?
text = data . decode ( 'utf-8' , errors = 'replace' )
# 或使用替代方案
text = data . decode ( 'utf-8' , errors = 'backslashreplace' )
UnicodeEncodeError
问题表现 :
1
UnicodeEncodeError : 'ascii' codec can 't encode character ' \u4e2d '
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
# 写入文件时指定编码
with open ( 'output.txt' , 'w' , encoding = 'utf-8' ) as f :
f . write ( chinese_text )
# 打印到控制台 (Windows)
import sys
import io
sys . stdout = io . TextIOWrapper ( sys . stdout . buffer , encoding = 'utf-8' )
# JSON 输出保留中文
import json
json . dumps ( data , ensure_ascii = False , indent = 2 )
JSON 解析失败
JSONDecodeError
问题表现 :
1
json . decoder . JSONDecodeError : Expecting value : line 1 column 1 ( char 0 )
原因分析 :
响应不是 JSON 格式 JSON 格式错误 空响应 HTML 错误页面 解决方案 :
1. 检查响应内容
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import requests
import json
response = requests . get ( url )
# 先检查状态码
if response . status_code != 200 :
print ( f "Error: { response . status_code } " )
print ( response . text )
exit ()
# 检查 Content-Type
content_type = response . headers . get ( 'Content-Type' , '' )
if 'application/json' not in content_type :
print ( f "Warning: Content-Type is { content_type } " )
# 安全解析
try :
data = response . json ()
except json . JSONDecodeError as e :
print ( f "JSON Parse Error: { e } " )
print ( f "Response text: { response . text [: 500 ] } " ) # 打印前500字符
data = None
2. 处理畸形 JSON
1
2
3
4
5
6
7
8
9
10
11
import json5 # pip install json5
# json5 可以解析带注释、尾随逗号的 JSON
text = '''
{
"name": "value", // 注释
"items": [1, 2, 3,], // 尾随逗号
}
'''
data = json5 . loads ( text )
3. 修复常见 JSON 问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import re
import json
def fix_json ( text ):
"""修复常见的 JSON 问题"""
# 移除 JavaScript 注释
text = re . sub ( r '//.*?\n' , ' \n ' , text )
text = re . sub ( r '/\*.*?\*/' , '' , text , flags = re . DOTALL )
# 移除尾随逗号
text = re . sub ( r ',\s*}' , '}' , text )
text = re . sub ( r ',\s*]' , ']' , text )
# 单引号转双引号 (谨慎使用)
# text = text.replace("'", '"')
return text
# 使用
text = response . text
fixed_text = fix_json ( text )
data = json . loads ( fixed_text )
4. 从 HTML 中提取 JSON
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import re
import json
from bs4 import BeautifulSoup
html = response . text
# 方法 1: 从 script 标签提取
soup = BeautifulSoup ( html , 'lxml' )
script = soup . find ( 'script' , { 'id' : 'initial-data' })
if script :
data = json . loads ( script . string )
# 方法 2: 使用正则提取
match = re . search ( r 'var\s+data\s*=\s*({.*?});' , html , re . DOTALL )
if match :
json_str = match . group ( 1 )
data = json . loads ( json_str )
# 方法 3: 提取 JSON-LD
script = soup . find ( 'script' , { 'type' : 'application/ld+json' })
if script :
data = json . loads ( script . string )
数据库连接问题
MongoDB 连接失败
问题表现 :
1
ServerSelectionTimeoutError : localhost : 27017 : [ Errno 111 ] Connection refused
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import pymongo
from pymongo.errors import ServerSelectionTimeoutError
# 添加超时和错误处理
try :
client = pymongo . MongoClient (
'mongodb://localhost:27017/' ,
serverSelectionTimeoutMS = 5000 ,
connectTimeoutMS = 5000 ,
socketTimeoutMS = 5000
)
# 测试连接
client . server_info ()
print ( "✅ MongoDB connected" )
except ServerSelectionTimeoutError as e :
print ( f "❌ MongoDB connection failed: { e } " )
print ( "提示:" )
print ( "1. 检查 MongoDB 是否运行: systemctl status mongod" )
print ( "2. 检查端口: netstat -an | grep 27017" )
print ( "3. 检查防火墙设置" )
exit ()
认证问题
1
2
3
4
5
6
7
8
9
10
# 带认证的连接
client = pymongo . MongoClient (
'mongodb://username:password@localhost:27017/' ,
authSource = 'admin' , # 认证数据库
authMechanism = 'SCRAM-SHA-256' # 认证机制
)
# 或使用 URI 格式
uri = 'mongodb://username:password@host1:27017,host2:27017/database?authSource=admin'
client = pymongo . MongoClient ( uri )
MySQL 连接问题
问题表现 :
1
Error 2003 : Can 't connect to MySQL server on ' localhost '
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import mysql.connector
from mysql.connector import Error
try :
connection = mysql . connector . connect (
host = 'localhost' ,
port = 3306 ,
database = 'testdb' ,
user = 'root' ,
password = 'password' ,
connect_timeout = 10 ,
autocommit = True ,
charset = 'utf8mb4'
)
if connection . is_connected ():
db_info = connection . get_server_info ()
print ( f "✅ Connected to MySQL Server version { db_info } " )
except Error as e :
print ( f "❌ Error: { e } " )
print ( "检查项:" )
print ( "1. MySQL 服务是否运行" )
print ( "2. 用户名密码是否正确" )
print ( "3. 是否允许远程连接" )
print ( "4. 防火墙设置" )
finally :
if connection and connection . is_connected ():
connection . close ()
Redis 连接问题
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import redis
from redis.exceptions import ConnectionError
try :
r = redis . Redis (
host = 'localhost' ,
port = 6379 ,
password = 'password' , # 如果有密码
db = 0 ,
decode_responses = True ,
socket_connect_timeout = 5 ,
socket_timeout = 5
)
# 测试连接
r . ping ()
print ( "✅ Redis connected" )
except ConnectionError as e :
print ( f "❌ Redis connection failed: { e } " )
except redis . AuthenticationError :
print ( "❌ Redis authentication failed" )
文件读写错误
文件不存在
问题表现 :
1
FileNotFoundError : [ Errno 2 ] No such file or directory : 'data.txt'
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import os
from pathlib import Path
# 方法 1: 检查文件是否存在
file_path = 'data.txt'
if os . path . exists ( file_path ):
with open ( file_path , 'r' ) as f :
content = f . read ()
else :
print ( f "File not found: { file_path } " )
# 方法 2: 使用 Path 对象
file_path = Path ( 'data.txt' )
if file_path . exists ():
content = file_path . read_text ( encoding = 'utf-8' )
# 方法 3: 创建目录
output_dir = Path ( 'output/data' )
output_dir . mkdir ( parents = True , exist_ok = True )
# 方法 4: 使用绝对路径
from pathlib import Path
base_dir = Path ( __file__ ) . resolve () . parent
file_path = base_dir / 'data' / 'file.txt'
权限错误
问题表现 :
1
PermissionError : [ Errno 13 ] Permission denied : '/var/log/app.log'
解决方案 :
1
2
3
4
5
6
7
8
9
10
11
12
# Linux/Mac
# 1. 修改文件权限
chmod 666 /var/log/app.log
# 2. 修改目录权限
chmod 777 /var/log/
# 3. 修改所有者
sudo chown $USER :$USER /var/log/app.log
# 4. 使用用户目录
# 在代码中使用 ~/data/ 而不是 /var/log/
1
2
3
4
5
6
7
8
# 使用用户目录
from pathlib import Path
home = Path . home ()
log_dir = home / '.myapp' / 'logs'
log_dir . mkdir ( parents = True , exist_ok = True )
log_file = log_dir / 'app.log'
内存问题
内存溢出
问题表现 :
解决方案 :
1. 分批处理大文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 不要一次性读取整个文件
# ❌ 错误方式
with open ( 'huge_file.txt' , 'r' ) as f :
content = f . read () # 可能导致内存溢出
# ✅ 正确方式:按行读取
with open ( 'huge_file.txt' , 'r' ) as f :
for line in f :
process_line ( line )
# 或使用生成器
def read_large_file ( file_path , chunk_size = 1024 * 1024 ):
"""分块读取大文件"""
with open ( file_path , 'r' ) as f :
while True :
chunk = f . read ( chunk_size )
if not chunk :
break
yield chunk
for chunk in read_large_file ( 'huge_file.txt' ):
process_chunk ( chunk )
2. 使用迭代器
1
2
3
4
5
6
7
8
9
10
# ❌ 一次性加载所有数据
urls = [ f "https://example.com/page/ { i } " for i in range ( 100000 )]
# ✅ 使用生成器
def url_generator ( count ):
for i in range ( count ):
yield f "https://example.com/page/ { i } "
for url in url_generator ( 100000 ):
process_url ( url )
3. 清理不用的对象
1
2
3
4
5
6
7
8
import gc
# 手动触发垃圾回收
gc . collect ()
# 删除大对象
del large_dataframe
gc . collect ()
4. 使用数据库而非内存
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ❌ 所有数据存在内存中
all_data = []
for item in items :
all_data . append ( process ( item ))
# ✅ 直接存入数据库
import sqlite3
conn = sqlite3 . connect ( 'data.db' )
cursor = conn . cursor ()
for item in items :
data = process ( item )
cursor . execute ( "INSERT INTO results VALUES (?, ?)" , ( data . id , data . value ))
conn . commit ()
CSV 处理问题
编码和分隔符
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import csv
import chardet
# 检测编码
with open ( 'data.csv' , 'rb' ) as f :
result = chardet . detect ( f . read ())
encoding = result [ 'encoding' ]
# 读取 CSV
with open ( 'data.csv' , 'r' , encoding = encoding ) as f :
# 自动检测分隔符
sample = f . read ( 1024 )
f . seek ( 0 )
sniffer = csv . Sniffer ()
delimiter = sniffer . sniff ( sample ) . delimiter
reader = csv . DictReader ( f , delimiter = delimiter )
for row in reader :
print ( row )
处理大型 CSV
1
2
3
4
5
6
7
8
9
10
11
12
import pandas as pd
# 分块读取
chunk_size = 10000
for chunk in pd . read_csv ( 'large.csv' , chunksize = chunk_size ):
process_chunk ( chunk )
# 只读取需要的列
df = pd . read_csv ( 'data.csv' , usecols = [ 'col1' , 'col2' ])
# 指定数据类型节省内存
df = pd . read_csv ( 'data.csv' , dtype = { 'id' : 'int32' , 'value' : 'float32' })
XML/HTML 解析问题
解析失败
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from bs4 import BeautifulSoup
from lxml import etree
html = response . text
# 方法 1: BeautifulSoup (宽容)
soup = BeautifulSoup ( html , 'lxml' ) # 或 'html.parser'
# 方法 2: lxml (严格)
try :
tree = etree . HTML ( html )
except etree . XMLSyntaxError as e :
print ( f "Parse error: { e } " )
# 处理畸形 HTML
from lxml.html import fromstring
from lxml.html.clean import Cleaner
cleaner = Cleaner ()
cleaned_html = cleaner . clean_html ( html )
doc = fromstring ( cleaned_html )
数据验证
使用 Pydantic
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from pydantic import BaseModel , validator , ValidationError
from typing import Optional
class Item ( BaseModel ):
id : int
name : str
price : float
stock : Optional [ int ] = 0
@validator ( 'price' )
def price_must_be_positive ( cls , v ):
if v <= 0 :
raise ValueError ( 'Price must be positive' )
return v
# 验证数据
try :
item = Item ( id = 1 , name = 'Product' , price = 19.99 )
except ValidationError as e :
print ( e . json ())
相关章节
相关内容