从巨大的 CSV 文件中读取随机行

Posted 2023-02-23

【中文标题】从巨大的 CSV 文件中读取随机行

【英文标题】：Read random lines from huge CSV file

【发布时间】：2012-06-04 21:39:14

【问题描述】：

我有一个非常大的 CSV 文件（15 Gb），我需要从中读取大约 100 万行随机行。 据我所见 - 并实现 - Python 中的 CSV 实用程序只允许在文件中按顺序迭代。

将所有文件读入内存以使用一些随机选择非常消耗内存，并且遍历所有文件并丢弃一些值并选择其他值非常耗时，所以有什么方法可以选择一些随机CSV 文件中的行并只读该行？

我试过没有成功：

import csv

with open('linear_e_LAN2A_F_0_435keV.csv') as file:
    reader = csv.reader(file)
    print reader[someRandomInteger]

CSV 文件示例：

331.093,329.735
251.188,249.994
374.468,373.782
295.643,295.159
83.9058,0
380.709,116.221
352.238,351.891
183.809,182.615
257.277,201.302
61.4598,40.7106

【问题讨论】：

可能重复***.com/questions/10605532/…

@VP 我相信那里的解决方案需要将整个文件加载到内存中。

@VP 不，在那个线程中，您在选择随机行之前将所有文件读入内存，这正是我需要避免的。

文件的格式是什么？所有的线都一样长吗？任何一行中是否有文字逗号？

@Andre 所有行都有 2 个十进制值，用逗号分隔。但是这些值可能有更多或更少的数字。我用文件样本更新了问题。

【参考方案1】：

import random

filesize = 1500                 #size of the really big file
offset = random.randrange(filesize)

f = open('really_big_file')
f.seek(offset)                  #go to random position
f.readline()                    # discard - bound to be partial line
random_line = f.readline()      # bingo!

# extra to handle last/first line edge cases
if len(random_line) == 0:       # we have hit the end
    f.seek(0)
    random_line = f.readline()  # so we'll grab the first line instead

正如@AndreBoos 所指出的，这种方法会导致选择有偏差。如果您知道行的最小和最大长度，则可以通过执行以下操作来消除此偏差：

假设（在这种情况下）我们有 min=3 和 max=15

1) 求上一行的长度（Lp）。

那么如果 Lp = 3，这条线最偏向于。因此，我们应该 100% 地使用它 如果 Lp = 15，则该线最偏向。我们应该只选择 20% 的时间，因为它被选中的可能性要高 5 倍。

我们通过随机保持行 X% 的时间来实现这一点：

X = min / Lp

如果我们不遵守规则，我们会再进行一次随机选择，直到我们的骰子结果正确为止。 :-)

【讨论】：

聪明，但这会在具有可变长度行的文件上提供有偏差的结果。

@thg435：这并不能解决任何问题。与短线之后的线相比，长线之后的线将不成比例地表示。此外，第一行将永远被读取。

@MariaZverina 我认为要进行公正的选择，您不仅需要知道行的最小长度，还需要知道文件中每行长度的频率。无论如何，在文件中，唯一少于 7 个字符的值是 0。我会尝试将我的文件转换为 0 替换为 0.00000，然后您的例程应该可以正常工作。

@MariaZverina，同意 jbssm 的评论。要纠正偏差，您需要知道线长的分布，然后相应地进行归一化。如果行长 3 ... 15 的分布是均匀的，则您的归一化技术有效，但要知道这种分布实际上是均匀的，我们需要读取整个文件。这让我们回到了同一个起点，如果我们要至少读取一次文件，为什么不把它写回将来更适合快速操作的东西（例如，在固定字段中或作为 sqlite3 表，等）

@MariaZverina ...好吧...我“算了算”，使用您的第二步过滤器，概率确实是一致的。我站得更正了。 （摘下帽子，向你鞠躬）

【参考方案2】：

我有一个非常大的 CSV 文件（15 Gb），我需要从中读取大约 100 万行随机行

假设您不需要恰好 100 万行并且事先知道 CSV 文件中的行数，您可以使用 reservoir sampling 来检索您的随机子集。只需遍历您的数据，并为每一行确定选择该行的机会。这样，您只需传递一次数据。

如果您需要经常提取随机样本但实际数据集不经常更改（因为您只需要在每次数据集更改时跟踪条目数），这很有效。

chances_selected = desired_num_results / total_entries
for line in csv.reader(file):
   if random() < chances_selected:
        result.append(line)

【讨论】：

是的，水库采样。但是他们如何找到total_entries？

@thg435 因此声明 "... 并知道 CSV 文件中的行数"。如果您经常进行采样并且只需要计算一次数据集大小，那么这种方案就可以工作。

@thg435 感谢您的恰当用语。我这辈子都想不起来了。

啊哈，没注意到。顺便说一句，如果所有行的长度都差不多，你可以估计total_entries = filesize / length_of_first_line

使用for line in CSV result 无论如何都会将文件加载到内存中，应该避免这种情况。

【参考方案3】：

您可以使用probabilistic method 的变体来选择文件中的随机行。

您可以保留一个大小为C 的缓冲区，而不是只保留一个被选中的数字。对于每个行号n，在具有N 行的文件中，您要选择概率为C/n（而不是原始1/n）的行。如果选择了编号，则选择随机位置从 C 长度缓冲区驱逐。

它是这样工作的：

import random

C = 2
fpath = 'somelines.txt'
buffer = []

f = open(fpath, 'r')
for line_num, line in enumerate(f):
    n = line_num + 1.0
    r = random.random()
    if n <= C:
        buffer.append(line.strip())
    elif r < C/n:
        loc = random.randint(0, C-1)
        buffer[loc] = line.strip()

这需要单次通过文件（因此它是线性时间）并从文件中返回准确 C 行。每行都有C/N被选中的概率。

为了验证上述方法是否有效，我创建了一个包含 5 行的文件，其中包含 a、b、c、d、e。我用 C=2 运行代码 10,000 次。这应该会产生大约 5 个选择 2（所以 10）个可能的选择的均匀分布。结果：

a,b: 1046
b,c: 1018
b,e: 1014
a,c: 1003
c,d: 1002
d,e: 1000
c,e: 993
a,e: 992
a,d: 985
b,d: 947

【讨论】：

嗨，谢谢。但是这样你必须先阅读所有文件，对于这个巨大的文件来说需要很多。

恭喜你一次搞定了如何做。

@jbssm 您必须至少通读整个文件一次才能获得公正的结果。

【参考方案4】：

如果您想多次抓取随机行（例如，用于机器学习的小批量），并且您不介意扫描一次大文件（不将其加载到内存中），那么您可以创建一个列表行 indeces 并使用 seek 快速抓取行（基于 Maria Zverina 的回答）。

# Overhead:
# Read the line locations into memory once.  (If the lines are long,
# this should take substantially less memory than the file itself.)
fname = 'big_file'
s = [0]
linelocs = [s.append(s[0]+len(n)) or s.pop(0) for n in open(fname)]
f = open(fname) # Reopen the file.

# Each subsequent iteration uses only the code below:
# Grab a 1,000,000 line sample
# I sorted these because I assume the seeks are faster that way.
chosen = sorted(random.sample(linelocs, 1000000))
sampleLines = []
for offset in chosen:
  f.seek(offset)
  sampleLines.append(f.readline())
# Now we can randomize if need be.
random.shuffle(sampleLines)

【讨论】：

刚刚注意到这是@parselmouth 已经描述的代码。

根据您的回答，我构建了一个小包，用于读取文件中的任意行。看看here

酷。我实际上再次需要这个，所以我会使用你的包。谢谢！

自从我上次发表评论以来，我用它做了一个 pip 包：random-access-file-reader

【参考方案5】：

如果这些行是真正的 .csv 格式并且不是固定字段，那么不，没有。您可以浏览文件一次，为每行索引字节偏移量，然后在以后需要时仅使用索引集，但无法先验预测任意 csv 文件的行终止 \n 字符的确切位置。

【讨论】：

我担心是这种情况，除非我们对每一行的可能值有更多了解。

@AndrewBuss 有，看看我的回答***.com/questions/10819911/…

【参考方案6】：

如果您知道总行数，则可以使用另一种解决方案 - 生成 100 万个随机数 (random.sample(xrange(n), 1000000))，直到总行数作为一组，然后使用：

for i, line in enumerate(csvfile):
    if i in lines_to_grab:
        yield line

这将以不偏不倚的方式准确地得到 100 万行，但您需要事先知道行数。

【讨论】：

但是这样我仍然必须阅读所有文件直到我真正想要阅读的行。

@jbssm：您必须遍历整个文件，是的，但您不必将其全部加载到内存中。

会起作用，但是会有一百万个数字的额外内存开销。而且您仍然需要通过数据文件。

是的。这在我的计算机上大约有 30MB，所以它可能不会破坏交易。

使用random.sample 创建lines_to_grab。

【参考方案7】：

如果您可以将此数据放在 sqlite3 数据库中，选择一些随机行数是微不足道的。您无需预先读取或填充文件中的行。由于 sqlite 数据文件是二进制的，因此您的数据文件将比 CSV 文本小 1/3 到 1/2。

您可以使用THIS 之类的脚本来导入 CSV 文件，或者更好的是，首先将您的数据写入数据库表。 SQLITE3 是 Python 发行版的一部分。

然后使用这些语句来获取 1,000,000 个随机行：

mydb='csv.db'
con=sqlite3.connect(mydb)

with con:
    cur=con.cursor()
    cur.execute("SELECT * FROM csv ORDER BY RANDOM() LIMIT 1000000;")

    for row in cur.fetchall():
        # now you have random rows...

【讨论】：

Than you Drew，这似乎是最先进的解决方案，但不幸的是，我不会是唯一一个使用这些数字来做科学的人，我真的很确定其他大多数人都不知道是什么是一个sql数据库，更不用说怎么用了。

【参考方案8】：

你可以用固定长度的记录重写文件，然后对中间文件进行随机访问：

ifile = file.open("inputfile.csv")
ofile = file.open("intermediatefile.csv",'w')
for line in ifile:
    ofile.write(line.rstrip('\n').ljust(15)+'\n')

那么，你可以这样做：

import random
ifile = file.open("intermediatefile.csv")
lines = []
samples = random.sample(range(nlines))
for sample in samples:
    ifile.seek(sample)
    lines.append(ifile.readline())

需要更多的磁盘空间，第一个程序可能需要一些时间才能运行，但它允许以后无限制地随机访问第二个程序的记录。

【讨论】：

我理解你的观点，它看起来有效，但你给我的转换程序不起作用。查看我的文件，所有值都有 7 个字符（包括 .），但 0 值除外。所以我实际上只需要将 0 转换为 0.00000 就可以了。

哦，我明白了；这就说得通了。您对输出该数据的程序有任何控制权吗？可能可以将输出格式更改为更常规的格式。

我愿意，但它是用 C++ 编写的，经过大量搜索后，我意识到 printf 无法为每个值做到这一点。所以我可能不得不在python中进行转换。

它应该是可行的尝试类似："%09.4f" % (1/3.0) .... 这将给你四个固定的小数点......并且不会换行下的任何数字10000.0

@jbssm 还考虑指数格式： "%.4e" % (1/3.0) 这将为您提供五个有效数字和 10 个字符的固定宽度。除非您还需要格式化负数 .... 在这种情况下，请使用 + 修饰符作为 printf。读取指数格式的浮点数和 Python 中的基本浮点数一样简单。

【参考方案9】：

# pass 1, count the number of rows in the file
rowcount = sum(1 for line in file)
# pass 2, select random lines
file.seek(0)
remaining = 1000000
for row in csv.reader(file):
    if random.randrange(rowcount) < remaining:
        print row
        remaining -= 1
    rowcount -= 1

【讨论】：

您仍然必须以这种方式遍历每一行才能到达随机变量告诉您​​读取的行。在考虑了更多之后，唯一可能不遍历所有文件的方法是使用 Maria Zverina 方法并确保所有行都具有相同数量的字符。

@jbssm，如果每行有相同数量的字符，这将变得微不足道 - 只需将随机行数乘以行大小，然后在文件中查找该点。

是的，我会写一些文件转换程序然后用那个方法。谢谢。

【参考方案10】：

在此方法中，我们生成一个随机数集，其元素数等于要读取的行数，其范围是数据中存在的行数。然后从最小到最大排序并存储。

然后逐行读取 csv 文件，并使用 line_counter 来表示行号。然后使用已排序随机数列表的第一个元素检查此line_counter，如果它们相同，则将该特定行写入新的 csv 文件中，并从列表中删除第一个元素，之前的第二个元素取而代之第一个，循环继续。

import random
k=random.sample(xrange(No_of_rows_in_data),No_of_lines_to_be_read)
Num=sorted(k)    
line_counter = 0

with open(input_file,'rb') as file_handle:
    reader = csv.reader(file_handle)
    with open(output_file,'wb') as outfile:
            a=csv.writer(outfile)
            for line in reader:
                line_counter += 1
                if line_counter == Num[0]:
                a.writerow(line)
                Num.remove(Num[0])
                if len(Num)==0:
                break    

【讨论】：

Python 依赖缩进；您在最内层循环 (for line in reader) 中的代码是 Python 缩进错误。尚不清楚Num.remove(Num[0]) 是否应该缩进——a.writerow(line) 必须缩进，break 也必须缩进。

【参考方案11】：

如果你可以使用pandas 和numpy，我已经发布了一个solution in another question，即pandas，具体但非常有效：

import pandas as pd
import numpy as np

filename = "data.csv"
sample_size = 1000000
batch_size = 5000

rng = np.random.default_rng()

sample_reader = pd.read_csv(filename, dtype=str, chunksize=batch_size)

sample = sample_reader.get_chunk(sample_size)

for chunk in sample_reader:
    chunk.index = rng.integers(sample_size, size=len(chunk))
    sample.loc[chunk.index] = chunk

更多详情，please see the other answer。

【参考方案12】：

def random_line(path, hint=1):
    with open(path, mode='rb') as file:
        import random
        while file.seek(random.randrange(file.seek(-2, 2))) and not file.readline(hint).endswith(b'\n'):
            pass
        return file.readline().decode().strip()

这是我为从一个非常大的文件中读取随机行而写的。

时间复杂度是O(k)，k是文本文件中行的平均长度。

hint 参数是文本文件中行的最小长度，如果你事先知道的话，用它来加速函数。

【参考方案13】：

总是为我工作

import csv
import random
randomINT = random.sample(range(1, 72655), 40000)
with open(file.csv,"rU") as fp:
    reader = csv.reader(fp, delimiter=",", quotechar='"', dialect=csv.excel_tab)
    data_read = [row for idx, row in enumerate(reader) if idx in randomINT]
    for idx, line in enumerate(data_read):
        pass