我有一个很大的合同CSV数据库(2500万行,大约7 GB).我需要过滤掉过期的合同,以减少进一步计算的大小.过期日期数组包含混合格式的日期(dd/mm/yyyy和dd/mm/yyyy hh/mm/ss).

我试过使用这个代码:

#reading database
reader = pd.read_csv(path_contracts, sep="|", header=0, low_memory=False, chunksize=1000000)   
output = "Base_Filtered.csv"
 
#filtering chunks
for contracts in reader:
        contracts[name_date_end] = pd.to_datetime(contracts[name_date_end], dayfirst=True, format='mixed')
        #conditional
        contracts = contracts[(contracts[name_date_end]>=date_report)]  
        #outputing filtered chunks     
        contracts.to_csv(output, sep="|")

但我得到了这个错误:

pandas._libs.tslibs.np_datetime.OutOfBoundsDatetime: Out of bounds nanosecond timestamp: 31.08.8020, at position 3584

我已经判断了位置3584,它返回06.11.2021:

print(contracts.at[3584,name_date_end])
06.11.2021

因此,我try 创建一个包含此元素的数据库的较小版本(5k行),它工作得非常好.

在try 过滤数据块之前,我曾try 一次过滤整个数据库.它产生了相同的误差,但在不同的位置.

找不到问题的根源和解决方案.

推荐答案

在我看来,你只需要两种日期或日期时间格式,而你的数据不止这些,但你不知道它们可能是什么.

对于如此大的数据集,很难事先对其进行询问并使一切完美,因此,您可能需要迭代和交互地运行此过程:

  • 日期时间与您的预定义格式不匹配的任何行都会被踢到单独的文件中
  • 然后,您可以判断这个"BADS"文件,并确定需要向您的主脚本添加什么新格式

我模拟了一个非常简单的输入CSV:

1,1/1/2001
2,2/2/2002 12/34/56
3,3.3.2003
4,6.1.2001
5,7.1.2001-5:38:19

以下脚本:

  • 定义预期的日期时间格式列表
  • 打开两个要写入的文件,并围绕它们创建csv.Writer
  • 开始读取输入CSV:
    • 它try 所有预定义的日期时间格式,直到成功并返回,或者
      • 如果不成功,则返回None
    • 如果上一步未成功,该行将写入Bad进行分析,否则...
    • 它根据正确的日期时间进行筛选,记录/打印过滤出的行(仅用于调试,在这个非常小的样例数据集上);它还将日期时间标准化为ISO格式
import csv
from datetime import datetime

fmts = [
    r"%d/%m/%Y",
    r"%d/%m/%Y %H/%M/%S",
]


def parse_dt(s: str) -> datetime | None:
    for fmt in fmts:
        try:
            dt = datetime.strptime(s, fmt)
            return dt
        except ValueError:
            continue

    return None


filtered = csv.writer(
    open("output_filtered.csv", "w", newline=""),
    delimiter=",",
)

bad = csv.writer(
    open("output_bad.csv", "w", newline=""),
    delimiter=",",
)

reader = csv.reader(
    open("input.csv", newline=""),
    delimiter=",",
)

for row in reader:
    dt = parse_dt(row[1])

    if dt is None:
        bad.writerow(row)
        continue

    if dt < datetime(2002, 1, 1):
        row[1] = dt.isoformat()  # standardize on ISO format
        filtered.writerow(row)
    else:
        print(f"dropped {dt}")

这将产生以下CSV:

output_bad.csv:
3,3.3.2003
4,6.1.2001
5,7.1.2001-5:38:19


output_filtered.csv:
1,2001-01-01T00:00:00

并将以下内容打印到stdout:

dropped 2002-02-02 12:34:56

然后我会 Select 3.3.20037.1.2001-5:38:19,用r"%d.%m.%Yr"%d.%m.%Y-%H:%M:%S"更新fmts:

fmts = [
    r"%d/%m/%Y",
    r"%d/%m/%Y %H/%M/%S",
    r"%d.%m.%Y",
    r"%d.%m.%Y-%H:%M:%S",
]

然后重新运行:

dropped 2002-02-02 12:34:56
dropped 2003-03-03 00:00:00

BADS CSV为空,过滤后的CSV如下所示:

1,2001-01-01T00:00:00
4,2001-01-06T00:00:00
5,2001-01-07T05:38:19

Python相关问答推荐

如何将新的SQL服务器功能映射到SQL Alchemy的ORM

PyQt5如何将pyuic 5生成的Python类添加到QStackedWidget中?

pyautogui.locateOnScreen在Linux上的工作方式有所不同

Python -Polars库中的滚动索引?

如何计算列表列行之间的公共元素

ModuleNotFound错误:没有名为flags.State的模块; flags不是包

NP.round解算数据后NP.unique

django禁止直接分配到多对多集合的前端.使用user.set()

如何创建一个缓冲区周围的一行与manim?

NumPy中条件嵌套for循环的向量化

driver. find_element无法通过class_name找到元素'""

将JSON对象转换为Dataframe

Python列表不会在条件while循环中正确随机化'

在pandas数据框中计算相对体积比指标,并添加指标值作为新列

AES—256—CBC加密在Python和PHP中返回不同的结果,HELPPP

用SymPy在Python中求解指数函数

Flask运行时无法在Python中打印到控制台

统计numpy. ndarray中的项目列表出现次数的最快方法

如何在Python中自动创建数字文件夹和正在进行的文件夹?

查找查找表中存在的列值组合