本教程旨在解决处理大型csv文件时常见的列数不一致和编码错误。我们将详细介绍如何利用python的`csv`模块，高效识别并报告csv文件中列数不符合预期标准的行，包括生成详细的单行报告和更简洁的行范围报告，并探讨如何正确处理unicode编码问题，确保数据导入前的质量检查。
在数据处理和导入（例如将Excel文件转换为CSV并上传至数据库如Teradata）的过程中，我们经常会遇到因数据录入不规范导致的CSV文件质量问题。其中最常见且棘手的包括：部分行的列数与其他行不一致，以及文件编码问题导致字符被错误解析（UnicodeDecodeError）。对于拥有数十万行和数十列的大型数据集，手动检查和修复是不可行的。本教程将指导您如何使用Python高效地识别并报告这些问题。
当面对一个包含125,000行、每行理论上应有66列的“脏数据”CSV文件时，简单的通过计数逗号来判断列数是远远不够的。这不仅容易受到数据中包含逗号的字段影响（如果字段未正确引用），更重要的是，它无法解决UnicodeDecodeError这类编码问题。
原始尝试：
with open('Data.csv', 'r') as csv_file:
for line in csv_file:
print( line.count(','))上述代码仅通过计数逗号来尝试识别列数，但它有几个主要缺陷：
立即学习“Python免费学习笔记（深入）”；
Python内置的csv模块是处理CSV文件的标准工具，它能够正确处理字段中的逗号、引号以及换行符，并且提供了灵活的编码处理机制。
UnicodeDecodeError通常是因为文件编码与读取时指定的编码不匹配。在open()函数中明确指定encoding参数是解决此问题的关键。常见的编码格式包括'utf-8'、'gbk'、'latin-1'等，具体取决于您的文件实际编码。
此外，使用csv.reader时，务必在open()函数中添加newline=''参数。这是因为csv.reader模块会自行处理行结束符，如果newline=''未指定，Python可能会在读取时错误地将换行符转换为\n，导致空行或字段解析错误。
此方法适用于需要详细了解每一行具体列数差异的场景。我们将遍历CSV文件的每一行，检查其列数是否与预期值（例如66列）匹配，并将不匹配的行号及其实际列数输出到报告文件。
示例数据 (input.csv)：
Col1,Col2,Col3 r1c1,r1c2 r2c1,r2c2,r2c3 r3c1 r4c1 r5c1 r6c1,r6c2,r6c3 r7c1,r7c2,r7c3 r8c1,r8c2 r9c1,r9c2
Python代码：
import csv
# 定义预期的列数
EXPECTED_COLS = 3 # 根据您的实际数据，这里应设置为66
# 打开输入和输出文件
# 务必指定正确的编码，例如 'utf-8' 或 'latin-1'
# newline='' 对于 csv 模块是必需的
try:
with open("input.csv", 'r', encoding='utf-8', newline='') as f_in, \
open("output_flat.csv", "w", encoding='utf-8', newline='') as f_out:
writer = csv.writer(f_out)
writer.writerow(["Row #", "N cols"]) # 写入报告头
reader = csv.reader(f_in)
# 跳过标题行（如果您的CSV文件有标题行）
# 如果没有标题行，请注释掉或移除下一行
try:
next(reader)
except StopIteration:
print("CSV文件为空或只有标题行。")
exit()
# 遍历每一行数据
for i, row in enumerate(reader, start=1):
# len(row) 返回当前行的列数
if len(row) != EXPECTED_COLS:
writer.writerow([i, len(row)])
print("列数不一致的行已报告至 output_flat.csv")
except FileNotFoundError:
print("错误：input.csv 文件未找到。请检查文件路径。")
except UnicodeDecodeError:
print("错误：解码文件时遇到问题。请尝试不同的编码，例如 'latin-1' 或 'gbk'。")
except Exception as e:
print(f"发生未知错误：{e}")
输出报告 (output_flat.csv) 示例：
Row #,N cols 1,2 3,1 4,1 5,1 8,2 9,2
这个报告清晰地列出了每一行（不含标题行）的行号以及其不符合预期的列数。
对于拥有大量不一致行的文件，逐行报告可能会生成一个非常大的报告文件。此时，将连续的、具有相同错误列数的行合并为行范围进行报告会更加简洁和实用。
示例数据 (input_large.csv)：
Col_1,Col_2,Col_3 r01c1,r01c2 r02c1,r02c2,r02c3 r03c1 r04c1 r05c1 r06c1,r06c2,r06c3 r07c1,r07c2,r07c3 r08c1,r08c2 r09c1,r09c2 r10c1,r10c2,r10c3 r11c1,r11c2,r11c3 r12c1,r12c2,r12c3 r13c1,r13c2,r13c3 r14c1,r14c2,r14c3 r15c1,r15c2,r15c3 r16c1 r17c1,r17c2 r18c1,r18c2 r19c1,r19c2 r20c1,r20c2 r21c1,r21c2 r22c1,r22c2,r22c3 r23c1,r23c2 r24c1,r24c2,r24c3 r25c1,r25c2 r26c1,r26c2,r26c3 r27c1,r27c2 r28c1,r28c2,r28c3 r29c1,r29c2 r30c1,r30c2 r31c1 r32c1,r32c2 r33c1 r34c1,r34c2,r34c3
Python代码：
import csv
# 定义预期的列数
EXPECTED_COLS = 3 # 根据您的实际数据，这里应设置为66
def write_range_row(writer_obj, col_count, row_start, row_end):
"""
将列数、起始行和结束行写入报告。
如果起始行和结束行相同，则只写入起始行。
"""
if row_start == row_end:
writer_obj.writerow([col_count, row_start, ""]) # 单行不一致
else:
writer_obj.writerow([col_count, row_start, row_end]) # 范围不一致
# 打开输入和输出文件
try:
with open("input_large.csv", 'r', encoding='utf-8', newline='') as f_in, \
open("output_ranges.csv", "w", encoding='utf-8', newline='') as f_out:
writer = csv.writer(f_out)
writer.writerow(["N cols", "Row start", "Row end"]) # 写入报告头
reader = csv.reader(f_in)
# 读取并跳过标题行，同时获取标题行的列数作为预期列数（如果标题行代表了正确的列数）
# 如果标题行不代表正确列数，请使用固定的 EXPECTED_COLS
try:
header_row = next(reader)
# 如果标题行的列数就是我们预期的正确列数，可以这样设置
# EXPECTED_COLS = len(header_row)
except StopIteration:
print("CSV文件为空或只有标题行。")
exit()
# 初始化跟踪变量
tracking = False
current_range_start_row = -1
current_range_cols_count = -1
# i 变量用于记录当前处理的数据行号（不含标题行）
i = 0
for i, row in enumerate(reader, start=1):
current_row_cols = len(row)
# 如果当前行的列数与预期列数不符
if current_row_cols != EXPECTED_COLS:
# 如果我们正在跟踪一个不一致的范围
if tracking:
# 如果当前行的列数与正在跟踪的范围列数不同，则结束前一个范围并开始新范围
if current_row_cols != current_range_cols_count:
write_range_row(writer, current_range_cols_count, current_range_start_row, i - 1)
current_range_start_row = i
current_range_cols_count = current_row_cols
else: # 否则，开始一个新的不一致范围的跟踪
tracking = True
current_range_start_row = i
current_range_cols_count = current_row_cols
else: # 如果当前行的列数与预期列数相符
# 如果我们正在跟踪一个不一致的范围，这意味着范围结束了
if tracking:
write_range_row(writer, current_range_cols_count, current_range_start_row, i - 1)
tracking = False
current_range_start_row = -1
current_range_cols_count = -1
# 循环结束后，检查是否还有未写入的跟踪范围
if tracking:
write_range_row(writer, current_range_cols_count, current_range_start_row, i)
print("列数不一致的行范围已报告至 output_ranges.csv")
except FileNotFoundError:
print("错误：input_large.csv 文件未找到。请检查文件路径。")
except UnicodeDecodeError:
print("错误：解码文件时遇到问题。请尝试不同的编码，例如 'latin-1' 或 'gbk'。")
except Exception as e:
print(f"发生未知错误：{e}")输出报告 (output_ranges.csv) 示例：
N cols,Row start,Row end 2,1, 1,3,5 2,8,9 1,16, 2,17,21 2,23, 2,25, 2,27, 2,29,30 1,31, 2,32, 1,33,
这个报告以更紧凑的方式展示了问题：例如，第3到5行都只有1列，第17到21行都只有2列。空“Row end”表示该行是单个不一致的行。
通过利用Python的csv模块，我们可以有效地识别和报告大型CSV文件中的列数不一致和编码问题。无论是需要详细的单行报告，还是更简洁的行范围报告，这些方法都能帮助您在数据导入数据库之前进行关键的数据质量检查。正确的编码处理和csv模块的规范使用是确保数据解析准确无误的基础，从而为后续的数据分析和存储奠定坚实的基础。
以上就是使用Python处理CSV文件中的列不一致及编码问题教程的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告
收藏
收藏
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
874
