如何用Python从SQL查询高效生成指定列宽的定长文件?
高效生成SQL Server查询结果的定长输出文件
需求说明
需要从SQL Server查询数据生成固定长度格式的文本文件,每行格式示例:
ABC11225_2023__Sacramento________John_________White_________05/13/1990
各列需严格遵循固定宽度,未填充部分用空格补全,超出部分截断,总长度188列,列宽要求如下:
| 字段名 | 宽度 |
|---|---|
| id | 9位 |
| year | 6位 |
| city | 20位 |
| firstname | 15位 |
| lastname | 12位 |
| birthdate | 10位 |
现有实现代码
已读取列宽配置文件columnWidths.csv,并通过pyodbc调用存储过程获取数据:
import csv import pyodbc # 读取列宽配置 file = open('columnWidths.csv', 'r') data = list(csv.reader(file, delimiter=",")) file.close() # 连接SQL Server并获取数据 server = "your_server_name" database = "your_db_name" cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=' + server+';Database='+database+';Trusted_Connection=yes;') cursor = cnxn.cursor() cursor.execute("{call [dbo].[GetApplicationData]}") applications = cursor.fetchall()
当前尝试的循环处理代码(存在性能顾虑):
for app in applications: a = "" for i in range(0, len(data)-1): columnWidth = int(data[i][0]) r = app[i] match r: case str(): a += app[i].ljust(columnWidth)[:columnWidth] case int(): a += str(app[i]).ljust(columnWidth)[:columnWidth] case date(): a += r.strftime("%m/%d/%Y").ljust(columnWidth)[:columnWidth] case None: a += ''.ljust(columnWidth)[:columnWidth] # 写入文件逻辑...
遇到的问题
每行数据可能存在NULL(对应Python的None)、需要类型转换(如布尔值Y转1、N转0),若为每个列单独用match:case处理类型与空值,担心处理大量数据时性能不足,希望找到更高效的实现方式。
高效实现方案
1. 预定义字段处理映射表
把每个字段的类型转换、格式化逻辑提前定义成可复用的处理函数,避免循环内重复判断类型,直接通过字段索引匹配处理逻辑:
from datetime import date # 读取列宽列表 column_widths = [int(row[0]) for row in data] # 动态生成字段处理函数(适配列宽配置) field_processors = [] for width in column_widths: def create_processor(w): def process(val): if val is None: return ' ' * w elif isinstance(val, date): return val.strftime("%m/%d/%Y").ljust(w)[:w] elif isinstance(val, bool): # 布尔值转换示例:True转1,False转0 return '1'.ljust(w)[:w] if val else '0'.ljust(w)[:w] elif isinstance(val, str) and val in ('Y', 'N'): # 字符串布尔值转换示例:Y转1,N转0 return '1'.ljust(w)[:w] if val == 'Y' else '0'.ljust(w)[:w] else: return str(val).ljust(w)[:w] return process field_processors.append(create_processor(width))
2. 优化字符串拼接与文件写入
避免循环内频繁拼接字符串(字符串是不可变类型,频繁拼接会生成大量中间对象),改用生成器表达式结合str.join()生成每行内容,同时直接逐行写入文件减少内存占用:
output_file = 'fixed_length_output.txt' with open(output_file, 'w', encoding='utf-8') as f: for app in applications: # 用生成器处理每个字段,一次性拼接成一行 line = ''.join(processor(val) for processor, val in zip(field_processors, app)) # 可选:校验行总长度是否符合要求 if len(line) != 188: # 此处可添加异常日志逻辑 pass f.write(line + '\n')
3. 数据库层预处理(可选)
如果数据量极大,可以在存储过程GetApplicationData中提前完成部分转换:
- 将NULL转换为空字符串或指定默认值
- 日期直接格式化为
mm/dd/yyyy格式的字符串 - 布尔值Y/N转换为1/0
这样Python端只需处理补宽和截断,减少内存占用和类型判断开销。
内容的提问来源于stack exchange,提问作者Connor Hale
相关产品推荐
相关产品推荐

