如何加快读取多个文件并将数据放入数据框中？

Question

我有许多文本文件，比如 50 个，我需要将它们读入一个庞大的数据框。目前，我正在使用以下步骤。

1. 阅读每个文件并检查标签是什么。我需要的信息通常包含在前几行中。相同的标签只是在文件的其余部分重复，每次都列出不同类型的数据。
2. 使用这些标签创建一个数据框。
3. 再次读取文件并用值填充数据框。
4. 将该数据帧与主数据帧连接起来。

这对于 100 KB 大小的文件非常有效 - 几分钟，但在 50 MB 时，只需要几个小时，而且不实用。

如何优化我的代码？特别是-

1. 如何确定哪些功能花费的时间最多，我需要优化哪些功能？是文件的读取吗？是写入数据框吗？我的计划在哪里花费时间？
2. 我应该考虑多线程还是多处理？
3. 我可以改进算法吗？
   • 也许将整个文件一次读入一个列表，而不是逐行读取，
   • 在块/整个文件中解析数据，而不是逐行解析，
   • 一次性将数据分配给数据帧，而不是逐行。
4. 我还能做些什么来让我的代码执行得更快吗？

这是一个示例代码。我自己的代码稍微复杂一些，因为文本文件更复杂，以至于我必须使用大约 10 个正则表达式和多个 while 循环来读取数据并将其分配到正确数组中的正确位置。为了保持 MWE 简单，我也没有在 MWE 的输入文件中使用重复标签，所以我希望我无缘无故地读取文件两次。我希望这是有道理的！

import re
import pandas as pd

df = pd.DataFrame()
paths = ["../gitignore/test1.txt", "../gitignore/test2.txt"]
reg_ex = re.compile('^(.+) (.+)\n')
# read all files to determine what indices are available
for path in paths:
    file_obj = open(path, 'r')
    print file_obj.readlines()

['a 1\n', 'b 2\n', 'end']
['c 3\n', 'd 4\n', 'end']

indices = []
for path in paths:
    index = []
    with open(path, 'r') as file_obj:
        line = True
        while line:
            try:
                line = file_obj.readline()
                match = reg_ex.match(line)
                index += match.group(1)
            except AttributeError:
                pass
    indices.append(index)
# read files again and put data into a master dataframe
for path, index in zip(paths, indices):
    subset_df = pd.DataFrame(index=index, columns=["Number"])
    with open(path, 'r') as file_obj:
        line = True
        while line:
            try:
                line = file_obj.readline()
                match = reg_ex.match(line)
                subset_df.loc[[match.group(1)]] = match.group(2)
            except AttributeError:
                pass
    df = pd.concat([df, subset_df]).sort_index()
print df

  Number
a      1
b      2
c      3
d      4

我的输入文件：

test1.txt

a 1
b 2
end

test2.txt

c 3
d 4
end

Answer 1

我已经多次使用它，因为它是多处理的一种特别简单的实现。

import pandas as pd
from multiprocessing import Pool

def reader(filename):
    return pd.read_excel(filename)

def main():
    pool = Pool(4) # number of cores you want to use
    file_list = [file1.xlsx, file2.xlsx, file3.xlsx, ...]
    df_list = pool.map(reader, file_list) #creates a list of the loaded df's
    df = pd.concat(df_list) # concatenates all the df's into a single df

if __name__ == '__main__':
    main()

使用它，您应该能够大大提高程序的速度，而无需进行太多工作。如果你不知道你有多少个处理器，你可以通过打开你的 shell 并输入来检查

echo %NUMBER_OF_PROCESSORS%

编辑：为了使运行更快，请考虑将文件更改为 csvs 并使用 pandas 函数pandas.read_csv

Answer 2

在拔出多处理锤之前，您的第一步应该是进行一些分析。使用 cProfile 快速浏览以确定哪些功能需要很长时间。不幸的是，如果您的行都在一个函数调用中，它们将显示为库调用。 line_profiler 更好，但需要更多的设置时间。

注意。如果使用 ipython，您可以使用 %timeit（timeit 模块的魔法命令）和 %prun（profile 模块的魔法命令）来为您的语句和函数计时。谷歌搜索将显示一些指南。

Pandas 是一个很棒的库，但我偶尔会因为使用不当而导致糟糕的结果。特别要注意 append()/concat() 操作。这可能是你的瓶颈，但你应该确定一下。通常，如果您不需要执行索引/列对齐，numpy.vstack() 和 numpy.hstack() 操作会更快。在您的情况下，您似乎可以使用可以节省时间的 Series 或 1-D numpy ndarrays。

顺便说一句，python 中的 try 块通常比检查无效条件慢 10 倍或更多，因此在将其插入每一行的循环时，请确保绝对需要它。这可能是另一个浪费时间的人。我想你在 match.group(1) 失败的情况下卡住了 try 块来检查 AttributeError 。我会先检查一个有效的匹配。

即使是这些小的修改也足以让您的程序在尝试像多处理这样剧烈的操作之前显着加快运行速度。这些 Python 库很棒，但也带来了一系列新的挑战。

Answer 3

首先，如果您要多次读取文件，这似乎会成为瓶颈。尝试将文件读入 1 个字符串对象，然后在其上多次使用 cStringIO。

其次，在读入所有文件之前，您并没有真正表明建立索引的任何理由。即使你这样做了，你为什么要使用 Pandas 进行 IO？似乎您可以在常规的 python 数据结构中构建它（可能使用__slots__），然后将其放入主数据框中。如果您在读取文件 Y 之前不需要文件 X 索引（正如您的第二个循环似乎建议的那样），您只需遍历文件一次。

第三，您可以在字符串上使用简单的split/strip 来提取空格分隔的标记，或者如果它更复杂（有字符串引号等），请使用 Python 标准库中的 CSV 模块.在您展示实际如何构建数据之前，很难提出与此相关的修复建议。

到目前为止，您所展示的内容可以通过简单的方式快速完成

for path in paths:
    data = []
    with open(path, 'r') as file_obj:
        for line in file_obj:
            try:
                d1, d2 = line.strip().split()
            except ValueError:
                pass
            data.append(d1, int(d2)))
    index, values = zip(*data)
    subset_df = pd.DataFrame({"Number": pd.Series(values, index=index)})

这是我在未预先分配磁盘空间的虚拟机上运行时的时间差异（生成的文件大小约为 24MB）：

import pandas as pd
from random import randint
from itertools import combinations
from posix import fsync


outfile = "indexValueInput"

for suffix in ('1', '2'):
    with open(outfile+"_" + suffix, 'w') as f:
        for i, label in enumerate(combinations([chr(i) for i in range(ord('a'), ord('z')+1)], 8)) :
            val = randint(1, 1000000)
            print >>f, "%s %d" % (''.join(label), val)
            if i > 3999999:
                break
        print >>f, "end"
        fsync(f.fileno())

def readWithPandas():
    data = []
    with open(outfile + "_2", 'r') as file_obj:
        for line in file_obj:
            try:
                d1, d2 = str.split(line.strip())
            except ValueError:
                pass
            data.append((d1, int(d2)))
    index, values = zip(*data)
    subset_df = pd.DataFrame({"Numbers": pd.Series(values, index=index)})

def readWithoutPandas():
    data = []
    with open(outfile+"_1", 'r') as file_obj:
        for line in file_obj:
            try:
                d1, d2 = str.split(line.strip())
            except ValueError:
                pass
            data.append((d1, int(d2)))
    index, values = zip(*data)

def time_func(func, *args):
    import time
    print "timing function", str(func.func_name)
    tStart = time.clock()
    func(*args)
    tEnd = time.clock()
    print "%f seconds " % (tEnd - tStart)

time_func(readWithoutPandas)
time_func(readWithPandas)

得到的时间是：

timing function readWithoutPandas
4.616853 seconds 
timing function readWithPandas
4.931765 seconds 

您可以在建立索引时尝试这些功能，看看时间上有什么不同。几乎可以肯定，减速来自多次磁盘读取。而且由于 Pandas 不会花时间从字典中构建您的数据框，因此您最好在将数据传递给 Pandas 之前弄清楚如何在纯 Python 中构建索引。但是在 1 个磁盘读取中同时读取数据和建立索引。

我想另一个警告是，如果您从代码内部打印，预计会花费大量时间。将纯文本写入 tty 所需的时间使读取/写入磁盘所需的时间相形见绌。

Answer 4

事实证明，首先创建一个空白 DataFrame，搜索索引以找到一行数据的正确位置，然后仅更新 DataFrame 的那一行是一个非常耗时的过程。

执行此操作的更快方法是将输入文件的内容读入原始数据结构，例如列表列表或字典列表，然后将其转换为 DataFrame。

当您读取的所有数据都在同一列中时，请使用列表。否则，使用 dicts 明确说明每个数据位应该去哪一列。

1 月 18 日更新：这是链接到How to parse complex text files using Python? 我还写了一个blog article explaining how to parse complex files to beginners。

Answer 5

python 的一般注意事项：

首先关于时间测量，您可以使用这样的 sn-p：

from time import time, sleep


class Timer(object):
    def __init__(self):
        self.last = time()


    def __call__(self):
        old = self.last
        self.last = time()
        return self.last - old

    @property
    def elapsed(self):
        return time() - self.last



timer = Timer()

sleep(2)
print timer.elapsed
print timer()
sleep(1)
print timer()

然后您可以多次对运行代码进行基准测试，并检查差异。

关于这个，我内联评论：

with open(path, 'r') as file_obj:
    line = True
    while line: #iterate on realdines instead.
        try:
            line = file_obj.readline()
            match = reg_ex.match(line)
            index += match.group(1)
            #if match:
            #    index.extend(match.group(1)) # or extend

        except AttributeError:
            pass

你之前的代码不是真正的 Pythonic，你可能想尝试/除外。 然后尝试只在尽可能少的行上做。

同样的注意事项也适用于第二个代码块。

如果您需要多次读取相同的文件。您可以使用 StringIO 将它们存储在 RAM 中，或者更轻松地保留一个您只读取一次的 {path: content} dict。

众所周知，Python 正则表达式很慢，您的数据看起来很简单，您可以考虑在输入行上使用 split 和 strip 方法。

 striped=[l.split() for l in [c.strip() for c in file_desc.readlines()] if l] 

我建议您阅读此内容：https://gist.github.com/JeffPaine/6213790 对应的视频在这里https://www.youtube.com/watch?v=OSGv2VnC0go

Answer 6

首先，为您的脚本使用分析器 (see this question)。准确分析哪个部分消耗更多时间。看看您是否可以优化它。

第二，我觉得I/O操作-文件读取最有可能是瓶颈。它可以使用并发方法进行优化。我建议同时读取文件并创建数据框。每个线程都可以将新创建​​的数据帧推送到队列中。主线程监控队列可以从队列中取出数据帧并与主数据帧合并。

希望这会有所帮助。

Answer 7

1 为文件创建一个输出模板（如结果数据框应该有 A、B C 列）

2 读取每个文件，将其转换为输出模板（在步骤 1 中建立）并保存文件，如 temp_idxx.csv，这可以并行完成 :)

3 将这些 temp_idxx.csv 文件拼接成一个大文件并删除 temps

这个程序的优点是它可以并行运行，而且不会吃掉所有的内存 缺点是创建输出格式并坚持使用它，以及磁盘空间使用情况

Answer 8

使用 pd.read_csv 将文件直接读入 pandas 数据帧。创建您的子集_df。使用诸如skipfooter之类的方法来跳过您知道不需要的文件末尾的行。还有更多方法可以替代您正在使用的一些正则表达式循环函数，例如 error_bad_lines 和 skip_blank_lines。

然后使用pandas提供的工具清理掉不需要的数据。

这将允许您只读取一次打开和读取文件。

Answer 9

您的代码不符合您的描述。

问题：1.阅读每个文件并检查标签是什么。我需要的信息通常包含在前几行中。

但是您阅读了整个文件，而不仅仅是几行。 这导致读取文件两次！

问题：2. 再次读取文件并用值填充数据框。

你在循环中一遍又一遍地覆盖df['a'|'b'|'c'|'d']，这是没用的
我相信这不是你想要的。
这适用于问题中给出的数据，但如果您必须处理 n 值则不行。

---

不同逻辑的提案：

data = {}
for path in paths:
    with open(path, 'r') as file_obj:
        line = True
        while line:
            try:
                line = file_obj.readline()
                match = reg_ex.match(line)
                if match.group(1) not in data:
                    data[ match.group(1) ] = []

                data[match.group(1)].append( match.group(2) )
            except AttributeError:
                pass

print('data=%s' % data)
df = pd.DataFrame.from_dict(data, orient='index').sort_index()
df.rename(index=str, columns={0: "Number"}, inplace=True)  

输出：

data={'b': ['2'], 'a': ['1'], 'd': ['4'], 'c': ['3']}
<class 'pandas.core.frame.DataFrame'>
Index: 4 entries, a to d
Data columns (total 1 columns):
Number    4 non-null object
dtypes: object(1)
memory usage: 32.0+ bytes
  Number
a      1
b      2
c      3
d      4  

时间表：

             Code from Q:   to_dict_from_dict
    4 values 0:00:00.033071 0:00:00.022146
 1000 values 0:00:08.267750 0:00:05.536500
10000 values 0:01:22.677500 0:00:55.365000

使用 Python：3.4.2 - pandas:0.19.2 - re:2.2.1 测试

Answer 10

您可以导入多处理模型并使用工作进程池同时打开多个文件作为文件对象，从而加快代码的加载部分。要测试时间，请导入 datetime 函数并使用以下代码：

import datetime
start=datetime.datetime.now()

#part of your code goes here

execTime1=datetime.datetime.now()
print(execTime1-start)

#the next part of your code goes here

execTime2=datetime.datetime.now()
print(execTime2-execTime1)

就每个文件只读取一次而言，考虑使用另一个多处理脚本来构建每个文件中的行列表，这样您就可以在不进行文件 I/O 操作的情况下检查匹配项。