我想在Python中将CSV加载到数据类中。数据类由字符串和枚举组成,我想相应地解析它。我现在知道有一个Python库可以做到这一点,但它不允许跳过格式错误的行,不幸的是,这些行是存在的。
我已经为此创建了一个方法,它可以读取文件,看起来像这样:
def dataset_reader(path: str):
#with open(path, 'r') as csv_handler:
reader = csv.reader(path)
header = reader.__next__()
expected_order = fields(MyFancyDataclass)
order_mapping = {fieldname: index for index, fieldname in enumerate([field.name for field in expected_order])}
header_mapping = {rowname: index for index, rowname in enumerate(header)}
order = [header_mapping.get(i[0]) for i in sorted(order_mapping.items(), key=lambda x: x[1])]
types = [type_ for type_ in [field.type for field in fields(MyFancyDataclass)]]
for line in reader:
try:
#yield MyFancyDataclass(*[line[x] if types[x] == str else types[x](line[x]) for x in order])
yield MyFancyDataclass(line[order[0]], line[order[1]], line[order[2]], line[order[3]], SourceType(line[order[4]]), line[order[5]], line[order[6]], line[order[7]],)
except Exception as e:
logging.error(line)
我基本上尝试做的是不假设CSV的写入顺序。只要所需的行在文件中,我们就解析它。为此,我首先读取头部,然后创建列Map的索引。然后,我对数据类执行相同的操作,并找到CSV的正确顺序。
然后我一行一行地读CSV,你会看到两种方法,一种是注解掉(注解掉更优雅,因为我们没有硬编码列数),另一种是更快。
我现在遇到的问题是,它仍然非常慢。当我们处理大数据时,这是一个小问题。对于如何加快速度,有什么好主意吗?Nogos正在假设CSV中的列顺序。尽管它应该始终保持相同的顺序,但我们不想假设它总是如此。因为本质上,所有事情都只是查找当前的产量。我看不出我们还可以改进什么来提高速度。
提前感谢所有的帮助!
用于复制的CSV文件,称之为test.csv:
key,value
123,aaa
234,bbb
12,aaa
1919191,bbb
12,
13,aaa
,bbb
,
123,bbb
用于复制的完整最小python脚本。将其存储在与test.csv相同的文件夹中:
from dataclasses import fields, dataclass
import logging
import csv
from enum import Enum
class SourceType(Enum):
a = "aaa"
b = "bbb"
@dataclass
class MyFancyDataclass:
key: str
value: SourceType
def dataset_reader(path: str):
#with open(path, 'r') as csv_handler:
reader = csv.reader(path)
header = reader.__next__()
expected_order = fields(MyFancyDataclass)
order_mapping = {fieldname: index for index, fieldname in enumerate([field.name for field in expected_order])}
header_mapping = {rowname: index for index, rowname in enumerate(header)}
order = [header_mapping.get(i[0]) for i in sorted(order_mapping.items(), key=lambda x: x[1])]
types = [type_ for type_ in [field.type for field in fields(MyFancyDataclass)]]
print(order)
for line in reader:
try:
#yield MyFancyDataclass(*[line[x] if types[x] == str else types[x](line[x]) for x in order])
yield MyFancyDataclass(line[order[0]], SourceType(line[order[1]]),)
except Exception as e:
print(e)
logging.error(line)
if __name__=="__main__":
print(list(dataset_reader(open("test.csv"))))
2条答案
按热度按时间fd3cxomn1#
我认为你会从csv的DictReader中得到很多好处,它掩盖了一致的订购问题:
接下来,我不知道您需要什么,因为您的示例数据为空值(
""
),但我不知道您如何处理数据类中必须接受字符串和枚举值的情况。我将提供以下代码来说明如何处理CSV中的空值(为了简洁起见,我还将枚举类重命名为
Value
):如果传入了一个非枚举字符串值(例如"ccc"),您想做什么?
我削减了您的示例CSV文件,命名为input-kv.csv:
并且还翻转了列input-vk. csv:
当我运行整个过程时:
我得到:
这并不完全准确,因为Fancy构造函数只接受字符串,但我认为它在翻转列和空白字段之间得到了相同的、一致的结果。
mfpqipee2#
如果速度是最重要的,那么你会希望在你的row-in-reader循环中削减尽可能多的动态性......至少我认为......我没有计时或分析这个......只是根据你自己的分析,DictReader太慢了,所以......
我能想到的处理任意顺序的列的最快方法是什么?显式命名您期望的列,然后根据标题获取它们的行索引:
否则,就像您之前那样,您将在row-in-reader循环中执行某种字段到列的Map查找,我发现这确实会降低数百万行的性能,但这也只是我的直觉/推测。
下面是我的完整测试程序(而且,我再次修改了您的值/语义,以尝试更好地说明问题的范围):
x一个一个一个一个x一个一个二个x
根据我对Cost of exception handlers的理解,我还将try/catch改为显式if/continue:IF块总体上更快,并且在预期的行将无效的情况下肯定更快。
您的团队必须保持数据类中的字段名和函数中的
idx_
变量同步,但这是为了在运行时提高速度而做出的权衡(在我看来)。无论如何,您似乎依赖于类型提示(也许是一个linter?),这将有助于捕捉不匹配。一些不匹配至少会导致运行时错误:而且,您可以只使用完整的列集进行当前单元测试。
我想出了一个方案,让文档字符串的字段名下面有与iter函数索引Map行中的列名匹配的列名,例如:
并提出了一个"linter"(我第一次尝试使用ast模块),它确保这些文档字符串与花哨的iter func中的字符串常量匹配:
只要你的数据类和iter函数是同步的:
linter很高兴,但是一旦linter在dataclass和iter func中都找不到任何列,或者发现两者不同步:
或
这两个错误都将在运行时引发异常:
或: