我有 2 个文本文件 (*.txt),其中包含以下格式的唯一字符串:
udtvbacfbbxfdffzpwsqzxyznecbqxgebuudzgzn:refmfxaawuuilznjrxuogrjqhlmhslkmprdxbascpoxda
ltswbjfsnejkaxyzwyjyfggjynndwkivegqdarjg:qyktyzugbgclpovyvmgtkihxqisuawesmcvsjzukcbrzi
第一个文件包含5000 万行 (4.3 GB),第二个文件包含100 万行 (112 MB)。一行包含 40 个字符、分隔符 : 和另外 45 个字符。
任务:获取两个文件的唯一值。也就是说,您需要一个csv 或 txt文件,其中的行位于第二个文件中,而哪些行不在第一个文件中。
我正在尝试使用vaex(Vaex)来做到这一点:
import vaex
base_files = ['file1.txt']
for i, txt_file in enumerate(base_files, 1):
for j, dv in enumerate(vaex.from_csv(txt_file, chunk_size=5_000_000, names=['data']), 1):
dv.export_hdf5(f'hdf5_base/base_{i:02}_{j:02}.hdf5')
check_files = ['file2.txt']
for i, txt_file in enumerate(check_files, 1):
for j, dv in enumerate(vaex.from_csv(txt_file, chunk_size=5_000_000, names=['data']), 1):
dv.export_hdf5(f'hdf5_check/check_{i:02}_{j:02}.hdf5')
dv_base = vaex.open('hdf5_base/*.hdf5')
dv_check = vaex.open('hdf5_check/*.hdf5')
dv_result = dv_check.join(dv_base, on='data', how='inner', inplace=True)
dv_result.export(path='result.csv')
结果,我得到了具有唯一行值的result.csv文件。但是验证过程需要很长时间。此外,它使用所有可用的 RAM 和所有处理器资源。如何加速这个过程?我究竟做错了什么?有什么可以做得更好?是否值得使用其他库(pandas,dask)进行此检查,它们会更快吗?
UPD 10.11.2020 到目前为止,我没有发现比以下选项更快的东西:
from io import StringIO
def read_lines(filename):
handle = StringIO(filename)
for line in handle:
yield line.rstrip('\n')
def read_in_chunks(file_obj, chunk_size=10485760):
while True:
data = file_obj.read(chunk_size)
if not data:
break
yield data
file_check = open('check.txt', 'r', errors='ignore').read()
check_set = {elem for elem in read_lines(file_check)}
with open(file='base.txt', mode='r', errors='ignore') as file_base:
for idx, chunk in enumerate(read_in_chunks(file_base), 1):
print(f'Checked [{idx}0 Mb]')
for elem in read_lines(chunk):
if elem in check_set:
check_set.remove(elem)
print(f'Unique rows: [{len(check_set)}]')
UPD 11.11.2020:感谢 @m9_psy 提供提高性能的提示。它真的更快!目前,最快的方法是:
from io import BytesIO
check_set = {elem for elem in BytesIO(open('check.txt', 'rb').read())}
with open('base.txt', 'rb') as file_base:
for line in file_base:
if line in check_set:
check_set.remove(line)
print(f'Unique rows: [{len(check_set)}]')
有没有办法进一步加快这个过程?