Linux 中的大型 CSV/文本处理答案

【问题标题】：Large CSV/text processing in LinuxLinux 中的大型 CSV/文本处理
【发布时间】：2020-06-27 00:56:39
【问题描述】：

我有一组相当大的 CSV 文件（每个约 1M 记录，15MB+），我需要提取其内容，重新格式化，然后附加到模板文本文件。

我编写了以下 python 脚本来执行此操作，它可以执行我想要的操作，但运行速度非常慢（单个文件的处理时间约为 15 分钟），并且我有数千个 CSV 文件要处理。

我写的脚本如下：

import pandas as pd
import shutil as s
from datetime import datetime as dt


dir = '/media/E/data/idb/'
f = '20020401.csv'
outFile = 'importbase.txt'


df = pd.read_csv(dir + f)
df["DataDate"] = pd.to_datetime(df["DataDate"]).view('int64') 
df["Expiration"] = pd.to_datetime(df["Expiration"], format='%m/%d/%Y')

df["measurement"] = ['options' for t in range(len(df))]
lines = [str(df["measurement"][d])
        + ",type=options"
        + " " 
        + "ticker=" + str(df["UnderlyingSymbol"][d])
        + ",symbol=" + str(df["OptionSymbol"][d])
        + ",expDate=" + str(df["Expiration"][d])
        + ",cont=" + str(df["Type"][d])
        + ",price=" + str(df["UnderlyingPrice"][d])
        + ",strike=" + str(df["Strike"][d])
        + ",last=" + str(df["Last"][d])
        + ",bid=" + str(df["Bid"][d])
        + ",ask=" + str(df["Ask"][d])
        + ",volume=" + str(df["Volume"][d])
        + ",OI=" + str(df["OpenInterest"][d])
        + ",IV=" + str(df["IV"][d])
        + ",delta=" + str(df["Delta"][d])
        + ",gamma=" + str(df["Gamma"][d])
        + ",theta=" + str(df["Theta"][d])
        + ",vega=" + str(df["Vega"][d])
        + ",aka=" + str(df["AKA"][d])
        + " " + str(df["DataDate"][d]) for d in range(len(df))]


a = s.copyfile(dir + outFile, dir + 'import.txt')               
file = open(a, 'a+')
for item in lines:
        file.write("%s\n" % item)
file.close()

importbase.txt 文件是模板文件。我需要将我的数据附加到其内容中，然后将其保存为import.txt 以进行处理。 importbase、import 和我的 CSV 示例如下：

importbase.txt

# DDL
CREATE DATABASE import

# DML
# CONTEXT-DATABASE: import

import.txt

# DDL
CREATE DATABASE import

# DML
# CONTEXT-DATABASE: import
options,type=options ticker=A,symbol=A020420C00015000,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=15.0,last=0.0,bid=21.0,ask=22.3,volume=0,OI=0,IV=0.6145,delta=0.9987,gamma=0.0,theta=0.6848,vega=0.0aka=ADC 1017619200000000000
options,type=options ticker=A,symbol=A020420P00015000,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=15.0,last=0.0,bid=0.0,ask=0.1,volume=0,OI=0,IV=0.5964,delta=0.0,gamma=0.0,theta=0.0,vega=0.0aka=APC 1017619200000000000
options,type=options ticker=A,symbol=A020420C00017500,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=17.5,last=0.0,bid=18.7,ask=19.5,volume=0,OI=0,IV=0.6145,delta=0.9987,gamma=0.0,theta=0.6371,vega=0.0aka=ADW 1017619200000000000
options,type=options ticker=A,symbol=A020420P00017500,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=17.5,last=0.0,bid=0.0,ask=0.1,volume=0,OI=0,IV=0.5964,delta=0.0,gamma=0.0,theta=0.0,vega=0.0aka=APW 1017619200000000000

20020401.csv

UnderlyingSymbol,UnderlyingPrice,Exchange,OptionSymbol,OptionExt,Type,Expiration,DataDate,Strike,Last,Bid,Ask,Volume,OpenInterest,IV,Delta,Gamma,Theta,Vega,AKA
A,36.53,*,A020420C00015000,,call,04/20/2002,04/01/2002,15,0,21,22.3,0,0,0.6145,0.9987,0,0.6848,0,ADC
A,36.53,*,A020420P00015000,,put,04/20/2002,04/01/2002,15,0,0,0.1,0,0,0.5964,0,0,0,0,APC
A,36.53,*,A020420C00017500,,call,04/20/2002,04/01/2002,17.5,0,18.7,19.5,0,0,0.6145,0.9987,0,0.6371,0,ADW
A,36.53,*,A020420P00017500,,put,04/20/2002,04/01/2002,17.5,0,0,0.1,0,0,0.5964,0,0,0,0,APW
A,36.53,*,A020420C00020000,,call,04/20/2002,04/01/2002,20,0,16.2,17,0,0,0.6145,0.9987,0,0.5882,0.0002,ADD
A,36.53,*,A020420P00020000,,put,04/20/2002,04/01/2002,20,0,0,0.1,0,0,0.5964,0,0,-0.0006,0.0001,APD
A,36.53,*,A020420C00022500,,call,04/20/2002,04/01/2002,22.5,5.7,13.7,14.5,0,5,0.6145,0.9985,0.0001,0.5092,0.0053,ADX
A,36.53,*,A020420P00022500,,put,04/20/2002,04/01/2002,22.5,0,0,0.1,0,1545,0.5964,-0.0001,0.0001,-0.0214,0.0036,APX
A,36.53,*,A020420C00025000,,call,04/20/2002,04/01/2002,25,11.6,11.2,12,0,545,0.6145,0.9963,0.0015,0.1189,0.0612,ADE
A,36.53,*,A020420P00025000,,put,04/20/2002,04/01/2002,25,0,0,0.1,0,882,0.5964,-0.0019,0.0012,-0.2882,0.0485,APE
A,36.53,*,A020420C00030000,,call,04/20/2002,04/01/2002,30,6.5,6.4,6.9,80,2357,0.6145,0.9314,0.0259,-6.1328,1.0651,ADF

我经常使用 bash，但想知道 sed 或 awk 之类的东西是否更适合处理时间可能成为问题的较低级别的任务。感谢任何有关方法的指导。

【问题讨论】：

您问的是/否问题，答案很简单。
看起来每行都有很好的条件处理，这意味着任何类型的操作系统级循环结构都会很慢（“Duh！”？）我可能会看@987654327 @ 解决方案，或者如果 *csv 文件有分隔符（例如，>>tabperl 解决方案可能很快（我很惊讶python 解决方案如此缓慢）
在您的代码和示例输出中，我在aka 之前缺少,：+ "aka=" + str(df["AKA"][d])。不是答案，但可能是一个错误。
是的，并行处理 CSV 文件会有所帮助，但目前约 15 分钟/文件的速度对于区区 1M 行/15MB 数据来说太慢了；最大的收益来自于将单个文件的处理时间降低到更合理的程度......然后添加并行处理将是最重要的:-)
@WalterA，你是对的，更新了。在这一点上没有引起问题，但是当我尝试使用它时会出现问题。谢谢

标签： python bash csv awk sed

【解决方案1】：

我不愿将此作为答案发布，因为这只是猜测。但是您可能会通过迭代数据框行并格式化单个字符串而不是连接一堆字符串来节省时间。

import pandas as pd
import shutil as s
from datetime import datetime as dt

dir = '/media/E/data/idb/'
f = '20020401.csv'
outFile = 'importbase.txt'


df = pd.read_csv(dir + f)
df["DataDate"] = pd.to_datetime(df["DataDate"]).view('int64') 
df["Expiration"] = pd.to_datetime(df["Expiration"], format='%m/%d/%Y')

df["measurement"] = ['options' for t in range(len(df))]

a = s.copyfile(dir + outFile, dir + 'import.txt')               
with open(a, 'a+') as file:
    for index, row in df.iterrows():
        file.write(
            "{measurement},type=options "
            "ticker={UnderlyingSymbol}"
            ",symbol={OptionSymbol}"
            ",expDate={Expiration}"
            ",cont={Type}"
            ",price={UnderlyingPrice}"
            ",strike={Strike}"
            ",last={Last}"
            ",bid={Bid}"
            ",ask={Ask}"
            ",volume={Volume}"
            ",OI={OpenInterest}"
            ",IV={IV}"
            ",delta={Delta}"
            ",gamma={Gamma}"
            ",theta={Theta}"
            ",vega={Vega}"
            ",aka={AKA}"
            " {DataDate}".format_map(row))

事实上，熊猫在这里并没有真正的优势。构建数据框会占用大量空间，但无论如何都需要通过 python 变量进行格式化，所以只需使用常规 CSV 模块

import csv
import shutil as s
from datetime import datetime as dt

dir = '/media/E/data/idb/'
f = '20020401.csv'
outFile = 'importbase.txt'

a = s.copyfile(dir + outFile, dir + 'import.txt')               

with open(dir + f) as in_fp, open(a, "a") as out_fp:
    reader = csv.DictReader(in_fp)
    for row in reader:
        # todo: figure out conversions outside pandas
        #row["DataDate"] = ?
        #row["Expiration"] = ? 
        row["measurement"] = "options"
        file.write(
            "{measurement},type=options "
            "ticker={UnderlyingSymbol}"
            ",symbol={OptionSymbol}"
            ",expDate={Expiration}"
            ",cont={Type}"
            ",price={UnderlyingPrice}"
            ",strike={Strike}"
            ",last={Last}"
            ",bid={Bid}"
            ",ask={Ask}"
            ",volume={Volume}"
            ",OI={OpenInterest}"
            ",IV={IV}"
            ",delta={Delta}"
            ",gamma={Gamma}"
            ",theta={Theta}"
            ",vega={Vega}"
            ",aka={AKA}"
            " {DataDate}".format_map(row))

【讨论】：

哇。概念上同意（只是昨晚写到“工作”），运行三个版本（我的原始版本，您的编辑，然后是第二次编辑）性能差异非常显着：0.5 分钟，0.45 分钟，0.02 分钟。（对我来说，昨晚我正在做其他处理，这导致原来的速度很慢）。
@Chris - 很高兴听到这 15 分钟的大部分时间是其他处理。我正在想象你将软盘放入奔腾 PC 中。我认为 csv 模块会更快，但这令人印象深刻。如果您永远不必单独触摸表格值，pandas 非常棒，但是一旦您这样做，您就需要支付 python 对象的构建价格。
不要以为您熟悉awk（或类似的低级别）并认为尝试使用它来改进事情还是值得的？我最终需要总共运行大约 4400 个文件。
@Chris 你应该可以通过使用 itertuples() 之类的东西来提高 Pandas 版本的性能，iterrows() 是出了名的迟钝。
事实上，我认为DataFrame.to_dict() 会更好。

【解决方案2】：

您可以使用 awk 来完成，但最好使用 CSV 解析器，而且由于您已经在使用 Python，因此更好地使用它可能就是答案。

您的问题的核心是将 CSV 复制为以逗号分隔的名称-值对列表。为此，这将完成这项工作，我希望它表现得很好。

#! /usr/bin/env python3
import csv

fieldnames = ('type', 'ticker', 'symbol', 'expDate', 'cont',
                  'price', 'strike', 'last', 'bid', 'ask', 'volume',
                  'OI', 'IV', 'delta', 'gamma',
                  'UnderlyingSymbol', 'UnderlyingPrice', 'Exchange',
                  'OptionSymbol', 'OptionExt', 'Type', 'Expiration', 'DataDate',
                  'Strike', 'Last', 'Bid', 'Ask', 'Volume',
                  'OpenInterest', 'IV', 'Delta', 'Gamma', 'Theta', 'Vega', 'AKA')

with open('input.csv', newline='') as csvfile:
    rows = csv.DictReader(csvfile, fieldnames)
    for i, row in enumerate(rows):
        if i == 0:
            continue
        line = ','.join( [ '%s=%s' % (k,v) for k,v in row.items() ] )
        print( 'options,%s' % line )

产生输出：

$ ./convert.py | nl
     1  options,type=    A,ticker=36.53,symbol=*,expDate=A020420P00015000,cont=,price=put,strike=04/20/2002,last=04/01/2002,bid=15,ask=0,volume=0,OI=0.1,IV=None,delta=0,gamma=0.5964,UnderlyingSymbol=0,UnderlyingPrice=0,Exchange=0,OptionSymbol=0,OptionExt=APC,Type=None,Expiration=None,DataDate=None,Strike=None,Last=None,Bid=None,Ask=None,Volume=None,OpenInterest=None,Delta=None,Gamma=None,Theta=None,Vega=None,AKA=None
     2  options,type=    A,ticker=36.53,symbol=*,expDate=A020420C00017500,cont=,price=call,strike=04/20/2002,last=04/01/2002,bid=17.5,ask=0,volume=18.7,OI=19.5,IV=None,delta=0,gamma=0.6145,UnderlyingSymbol=0.9987,UnderlyingPrice=0,Exchange=0.6371,OptionSymbol=0,OptionExt=ADW,Type=None,Expiration=None,DataDate=None,Strike=None,Last=None,Bid=None,Ask=None,Volume=None,OpenInterest=None,Delta=None,Gamma=None,Theta=None,Vega=None,AKA=None
     3  options,type=    A,ticker=36.53,symbol=*,expDate=A020420P00017500,cont=,price=put,strike=04/20/2002,last=04/01/2002,bid=17.5,ask=0,volume=0,OI=0.1,IV=None,delta=0,gamma=0.5964,UnderlyingSymbol=0,UnderlyingPrice=0,Exchange=0,OptionSymbol=0,OptionExt=APW,Type=None,Expiration=None,DataDate=None,Strike=None,Last=None,Bid=None,Ask=None,Volume=None,OpenInterest=None,Delta=None,Gamma=None,Theta=None,Vega=None,AKA=None
     4  options,type=    A,ticker=36.53,symbol=*,expDate=A020420C00020000,cont=,price=call,strike=04/20/2002,last=04/01/2002,bid=20,ask=0,volume=16.2,OI=17,IV=None,delta=0,gamma=0.6145,UnderlyingSymbol=0.9987,UnderlyingPrice=0,Exchange=0.5882,OptionSymbol=0.0002,OptionExt=ADD,Type=None,Expiration=None,DataDate=None,Strike=None,Last=None,Bid=None,Ask=None,Volume=None,OpenInterest=None,Delta=None,Gamma=None,Theta=None,Vega=None,AKA=None

【讨论】：

嗨，我已经编辑以将输入作为上面的文本包含在内。输出中ticker 之前没有逗号是故意的，不是错误。
您的“CSV 文件”没有逗号吗？它是如何划分的？（我看到现在需要名称字典。您可以通过向 csv.DictReader 提供 fieldnames 参数来获得相同的效果。）
不确定我是否关注。逗号包含在上面的输入文件示例中。
明白了。我想我的页面刷新问题。我更新了我的答案以匹配。希望对您有所帮助。
@Chris 输出中ticker 之前没有逗号是故意的，而不是错误。这很奇怪，读取这些文件会是什么？

【解决方案3】：

以下是如何在 awk 中进行转换（使用 GNU awk 进行 mktime()）：

$ cat tst.awk
BEGIN {
    n = split("                                         \
                ticker          UnderlyingSymbol        \
                symbol          OptionSymbol            \
                expDate         Expiration              \
                cont            Type                    \
                price           UnderlyingPrice         \
                strike          Strike                  \
                last            Last                    \
                bid             Bid                     \
                ask             Ask                     \
                volume          Volume                  \
                OI              OpenInterest            \
                IV              IV                      \
                delta           Delta                   \
                gamma           Gamma                   \
                theta           Theta                   \
                vega            Vega                    \
                aka             AKA                     \
                DataDate        DataDate                \
        ", outInNameMap)

    for (i=1; i<n; i+=2) {
        outFldName = outInNameMap[i]
        inFldName  = outInNameMap[i+1]
        inNames2outNrs[inFldName]   = ++numOutFlds
        outNrs2outNames[numOutFlds] = outFldName
    }

    dateFmt["expDate"]  = "iso8601"
    dateFmt["DataDate"] = "epoch"

    FS=OFS=","
}

FNR == 1 {
    for (inFldNr=1; inFldNr<=NF; inFldNr++) {
        inFldName = $inFldNr
        outFldNr  = inNames2outNrs[inFldName]
        outNrs2inNrs[outFldNr] = inFldNr
    }
    next
}

{
    printf "options,type=options"
    for (outFldNr=1; outFldNr<=numOutFlds; outFldNr++) {
        outFldName = outNrs2outNames[outFldNr]
        inFldNr = outNrs2inNrs[outFldNr]
        fldVal  = $inFldNr
        if ( outFldName in dateFmt) {
            split(fldVal,d,"/")
            if ( dateFmt[outFldName] == "iso8601" ) {
                fldVal = sprintf("%s-%s-%s 00:00:00", d[3], d[1], d[2])
            }
            else if ( dateFmt[outFldName] == "epoch" ) {
                fldVal = mktime(sprintf("%s %s %s 00 00 00", d[3], d[1], d[2]))
            }
        }
        pfx = (outFldNr==numOutFlds ? " " : (outFldNr>1 ? OFS : " ") outFldName "=")
        printf "%s%s", pfx, fldVal
    }
    print ""
}

$ awk -f tst.awk file.csv
options,type=options ticker=A,symbol=A020420C00015000,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=15,last=0,bid=21,ask=22.3,volume=0,OI=0,IV=0.6145,delta=0.9987,gamma=0,theta=0.6848,vega=0,aka=ADC 1017640800
options,type=options ticker=A,symbol=A020420P00015000,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=15,last=0,bid=0,ask=0.1,volume=0,OI=0,IV=0.5964,delta=0,gamma=0,theta=0,vega=0,aka=APC 1017640800
options,type=options ticker=A,symbol=A020420C00017500,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=17.5,last=0,bid=18.7,ask=19.5,volume=0,OI=0,IV=0.6145,delta=0.9987,gamma=0,theta=0.6371,vega=0,aka=ADW 1017640800
options,type=options ticker=A,symbol=A020420P00017500,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=17.5,last=0,bid=0,ask=0.1,volume=0,OI=0,IV=0.5964,delta=0,gamma=0,theta=0,vega=0,aka=APW 1017640800
options,type=options ticker=A,symbol=A020420C00020000,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=20,last=0,bid=16.2,ask=17,volume=0,OI=0,IV=0.6145,delta=0.9987,gamma=0,theta=0.5882,vega=0.0002,aka=ADD 1017640800
options,type=options ticker=A,symbol=A020420P00020000,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=20,last=0,bid=0,ask=0.1,volume=0,OI=0,IV=0.5964,delta=0,gamma=0,theta=-0.0006,vega=0.0001,aka=APD 1017640800
options,type=options ticker=A,symbol=A020420C00022500,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=22.5,last=5.7,bid=13.7,ask=14.5,volume=0,OI=5,IV=0.6145,delta=0.9985,gamma=0.0001,theta=0.5092,vega=0.0053,aka=ADX 1017640800
options,type=options ticker=A,symbol=A020420P00022500,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=22.5,last=0,bid=0,ask=0.1,volume=0,OI=1545,IV=0.5964,delta=-0.0001,gamma=0.0001,theta=-0.0214,vega=0.0036,aka=APX 1017640800
options,type=options ticker=A,symbol=A020420C00025000,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=25,last=11.6,bid=11.2,ask=12,volume=0,OI=545,IV=0.6145,delta=0.9963,gamma=0.0015,theta=0.1189,vega=0.0612,aka=ADE 1017640800
options,type=options ticker=A,symbol=A020420P00025000,expDate=2002-04-20 00:00:00,cont=put,price=36.53,strike=25,last=0,bid=0,ask=0.1,volume=0,OI=882,IV=0.5964,delta=-0.0019,gamma=0.0012,theta=-0.2882,vega=0.0485,aka=APE 1017640800
options,type=options ticker=A,symbol=A020420C00030000,expDate=2002-04-20 00:00:00,cont=call,price=36.53,strike=30,last=6.5,bid=6.4,ask=6.9,volume=80,OI=2357,IV=0.6145,delta=0.9314,gamma=0.0259,theta=-6.1328,vega=1.0651,aka=ADF 1017640800

我不确定你的输出行上的最终值是多少，它似乎是自纪元以来的秒数变化，但我会将它作为练习留给你，以便你在必要时修复它。我使用了比必要更多的临时变量，并为所有这些变量使用了清晰的名称，因此发生的事情很明显，因此不需要评论来解释它，您应该能够轻松地在它的基础上进行构建。

【讨论】：