python - 在 python 中读取 15 M 行 csv 文件的有效方法

Question

对于我的应用程序，我需要读取多个文件，每个文件有 15 M 行，将它们存储在 DataFrame 中，并将 DataFrame 保存为 HDFS5 格式。

我已经尝试过不同的方法，特别是具有 chunksize 和 dtype 规范的 pandas.read_csv，以及 dask.dataframe。他们都需要大约 90 秒来处理 1 个文件，所以我想知道是否有一种方法可以按照描述的方式有效地处理这些文件。在下文中，我展示了一些我已经完成的测试的代码。

import pandas as pd
import dask.dataframe as dd
import numpy as np
import re 

# First approach
store = pd.HDFStore('files_DFs.h5')

chunk_size = 1e6

df_chunk = pd.read_csv(file,
                sep="\t",
                chunksize=chunk_size,
                usecols=['a', 'b'],
                converters={"a": lambda x: np.float32(re.sub(r"[^\d.]", "", x)),\
                            "b": lambda x: np.float32(re.sub(r"[^\d.]", "", x))},
                skiprows=15
           )              
chunk_list = [] 


for chunk in df_chunk:
      chunk_list.append(chunk)


df = pd.concat(chunk_list, ignore_index=True)

store[dfname] = df
store.close()

# Second approach

df = dd.read_csv(
        file,
        sep="\t",
        usecols=['a', 'b'],
        converters={"a": lambda x: np.float32(re.sub(r"[^\d.]", "", x)),\
                    "b": lambda x: np.float32(re.sub(r"[^\d.]", "", x))},
        skiprows=15
     )
store.put(dfname, df.compute())
store.close()

这是文件的样子(空白由文字制表符组成):

a   b
599.998413  14.142895
599.998413  20.105534
599.998413  6.553850
599.998474  27.116098
599.998474  13.060312
599.998474  13.766775
599.998596  1.826706
599.998596  18.275938
599.998718  20.797491
599.998718  6.132450)
599.998718  41.646194
599.998779  19.145775

Answer 1

首先回答问题

1- 如何高效读取包含 float 的 15M 行的 csv

我建议你使用 modin :

生成示例数据:

import modin.pandas as mpd
import pandas as pd
import numpy as np

frame_data = np.random.randint(0, 10_000_000, size=(15_000_000, 2)) 
pd.DataFrame(frame_data*0.0001).to_csv('15mil.csv', header=False)

!wc 15mil*.csv ; du -h 15mil*.csv

    15000000   15000000  480696661 15mil.csv
    459M    15mil.csv

现在是基准测试:

%%timeit -r 3 -n 1 -t
global df1
df1 = pd.read_csv('15mil.csv', header=None)
    9.7 s ± 95.1 ms per loop (mean ± std. dev. of 3 runs, 1 loop each)

%%timeit -r 3 -n 1 -t
global df2
df2 = mpd.read_csv('15mil.csv', header=None)
    3.07 s ± 685 ms per loop (mean ± std. dev. of 3 runs, 1 loop each)

(df2.values == df1.values).all()
    True

因此我们可以看到，modin 在我的设置中大约快 3 倍。

---

现在回答你的具体问题

2- 清理包含非数字字符的 csv 文件，然后读取它

正如人们所指出的，您的瓶颈可能是转换器。您正在调用这些 lambdas 3000 万次。在这种规模下，即使是函数调用开销也变得非常重要。

让我们来解决这个问题。

生成脏数据集:

!sed 's/.\{4\}/&)/g' 15mil.csv > 15mil_dirty.csv

方法

首先，我尝试将 modin 与 converters 参数一起使用。然后，我尝试了一种不同的方法来减少调用正则表达式的次数:

首先，我将创建一个类文件对象，通过您的正则表达式过滤所有内容:

class FilterFile():
    def __init__(self, file):
        self.file = file
    def read(self, n):
        return re.sub(r"[^\d.,\n]", "", self.file.read(n))
    def write(self, *a): return self.file.write(*a) # needed to trick pandas
    def __iter__(self, *a): return self.file.__iter__(*a) # needed

然后我们将它作为 read_csv 中的第一个参数传递给 pandas:

with open('15mil_dirty.csv') as file:
    df2 = pd.read_csv(FilterFile(file))

基准:

%%timeit -r 1 -n 1 -t
global df1
df1 = pd.read_csv('15mil_dirty.csv', header=None,
        converters={0: lambda x: np.float32(re.sub(r"[^\d.]", "", x)),
                    1: lambda x: np.float32(re.sub(r"[^\d.]", "", x))}
           )
    2min 28s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each)

%%timeit -r 1 -n 1 -t
global df2
df2 = mpd.read_csv('15mil_dirty.csv', header=None,
        converters={0: lambda x: np.float32(re.sub(r"[^\d.]", "", x)),
                    1: lambda x: np.float32(re.sub(r"[^\d.]", "", x))}
           )
    38.8 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each)

%%timeit -r 1 -n 1 -t
global df3
df3 = pd.read_csv(FilterFile(open('15mil_dirty.csv')), header=None,)
    1min ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each)

看来 modin 又赢了! 不幸的是，modin 还没有实现从缓冲区读取，所以我设计了最终的方法。

终极方法:

%%timeit -r 1 -n 1 -t
with open('15mil_dirty.csv') as f, open('/dev/shm/tmp_file', 'w') as tmp:
    tmp.write(f.read().translate({ord(i):None for i in '()'}))
df4 = mpd.read_csv('/dev/shm/tmp_file', header=None)
    5.68 s ± 0 ns per loop (mean ± std. dev. of 1 run, 1 loop each)

这使用 translate 比 re.sub 快得多，还使用 ​​/dev/shm 这是内存文件系统ubuntu(和其他 linux)通常提供。任何写在那里的文件永远不会进入磁盘，所以速度很快。 最后，它使用 modin 读取文件，绕过 modin 的缓冲区限制。 这种方法比您的方法快 30 倍，而且非常简单。