python - 通过 readlines(size) 提高大文件搜索效率
python - increase efficiency of large-file search by readlines(size)
我是 Python 的新手，我目前正在使用 Python 2. 我有一些源文件，每个文件都包含大量数据（大约 1900 万行）。看起来像下面这样：
apple \t N \t apple
n&apos
garden \t N \t garden
b\ta\md
great \t Adj \t great
nice \t Adj \t (unknown)
etc
我的任务是在每个文件的第3列中搜索一些目标词，每次在语料库中找到一个目标词，就必须将这个词前后的10个词添加到一个多维词典中。
编辑： 应排除包含“&”、“\”或字符串“(unknown)”的行。
我尝试使用 readlines() 和 enumerate() 来解决这个问题，如下面的代码所示。代码做了它应该做的，但对于源文件中提供的数据量来说，它显然不够高效。
我知道 readlines() 或 read() 不应该用于巨大的数据集，因为它将整个文件加载到内存中。尽管如此，逐行读取文件，我并没有设法使用枚举方法来获取目标词前后的10个词。
我也不能使用 mmap，因为我没有在该文件上使用它的权限。
所以，我想具有一些大小限制的 readlines 方法将是最有效的解决方案。但是，为此，我会不会犯一些错误，因为每次到达大小限制的末尾时，目标词之后的 10 个词都不会被捕获，因为代码刚刚中断？
def get_target_to_dict(file):
targets_dict = {}
with open(file) as f:
for line in f:
targets_dict[line.strip()] = {}
return targets_dict
targets_dict = get_target_to_dict('targets_uniq.txt')
# browse directory and process each file
# find the target words to include the 10 words before and after to the dictionary
# exclude lines starting with <,-,; to just have raw text
def get_co_occurence(path_file_dir, targets, results):
lines = []
for file in os.listdir(path_file_dir):
if file.startswith('corpus'):
path_file = os.path.join(path_file_dir, file)
with gzip.open(path_file) as corpusfile:
# PROBLEMATIC CODE HERE
# lines = corpusfile.readlines()
for line in corpusfile:
if re.match('[A-Z]|[a-z]', line):
if '(unknown)' in line:
continue
elif '\' in line:
continue
elif '&' in line:
continue
lines.append(line)
for i, line in enumerate(lines):
line = line.strip()
if re.match('[A-Z][a-z]', line):
parts = line.split('\t')
lemma = parts[2]
if lemma in targets:
pos = parts[1]
if pos not in targets[lemma]:
targets[lemma][pos] = {}
counts = targets[lemma][pos]
context = []
# look at 10 previous lines
for j in range(max(0, i-10), i):
context.append(lines[j])
# look at the next 10 lines
for j in range(i+1, min(i+11, len(lines))):
context.append(lines[j])
# END OF PROBLEMATIC CODE
for context_line in context:
context_line = context_line.strip()
parts_context = context_line.split('\t')
context_lemma = parts_context[2]
if context_lemma not in counts:
counts[context_lemma] = {}
context_pos = parts_context[1]
if context_pos not in counts[context_lemma]:
counts[context_lemma][context_pos] = 0
counts[context_lemma][context_pos] += 1
csvwriter = csv.writer(results, delimiter='\t')
for k,v in targets.iteritems():
for k2,v2 in v.iteritems():
for k3,v3 in v2.iteritems():
for k4,v4 in v3.iteritems():
csvwriter.writerow([str(k), str(k2), str(k3), str(k4), str(v4)])
#print(str(k) + "\t" + str(k2) + "\t" + str(k3) + "\t" + str(k4) + "\t" + str(v4))
results = open('results_corpus.csv', 'wb')
word_occurrence = get_co_occurence(path_file_dir, targets_dict, results)
出于完整性原因，我复制了整个代码部分，因为它是一个函数的全部部分，该函数根据提取的所有信息创建多维字典，然后将其写入 csv 文件。
如果有任何提示或建议可以提高此代码的效率，我将不胜感激。
编辑 我更正了代码，以便它考虑到目标词前后的确切 10 个词
我的想法是在10行之前创建一个缓冲区，在10行之后创建一个缓冲区，当文件被读取时，它将被推入缓冲区之前，如果大小超过10，缓冲区将被弹出
对于后缓冲区，我从文件迭代器 1st 克隆了另一个迭代器。然后 运行 两个迭代器在循环中与克隆迭代器并行 运行 提前 10 次迭代以获得后 10 行。
这样可以避免使用 readlines() 并将整个文件加载到内存中。
希望在实际案例中对你有用
已编辑：
如果第 3 列不包含“&”、“\”、“（未知）”中的任何一个，则仅填充前后缓冲区。还将 split('\t') 更改为 split() 以便它处理所有空格或选项卡
import itertools
def get_co_occurence(path_file_dir, targets, results):
excluded_words = ['&', '\', '(unknown)'] # modify excluded words here
for file in os.listdir(path_file_dir):
if file.startswith('testset'):
path_file = os.path.join(path_file_dir, file)
with open(path_file) as corpusfile:
# CHANGED CODE HERE
before_buf = [] # buffer to store before 10 lines
after_buf = [] # buffer to store after 10 lines
corpusfile, corpusfile_clone = itertools.tee(corpusfile) # clone file iterator to access next 10 lines
for line in corpusfile:
line = line.strip()
if re.match('[A-Z]|[a-z]', line):
parts = line.split()
lemma = parts[2]
# before buffer handling, fill buffer excluded line contains any of excluded words
if not any(w in line for w in excluded_words):
before_buf.append(line) # append to before buffer
if len(before_buf)>11:
before_buf.pop(0) # keep the buffer at size 10
# next buffer handling
while len(after_buf)<=10:
try:
after = next(corpusfile_clone) # advance 1 iterator
after_lemma = ''
after_tmp = after.split()
if re.match('[A-Z]|[a-z]', after) and len(after_tmp)>2:
after_lemma = after_tmp[2]
except StopIteration:
break # copy iterator will exhaust 1st coz its 10 iteration ahead
if after_lemma and not any(w in after for w in excluded_words):
after_buf.append(after) # append to buffer
# print 'after',z,after, ' - ',after_lemma
if (after_buf and line in after_buf[0]):
after_buf.pop(0) # pop off one ready for next
if lemma in targets:
pos = parts[1]
if pos not in targets[lemma]:
targets[lemma][pos] = {}
counts = targets[lemma][pos]
# context = []
# look at 10 previous lines
context= before_buf[:-1] # minus out current line
# look at the next 10 lines
context.extend(after_buf)
# END OF CHANGED CODE
# CONTINUE YOUR STUFF HERE WITH CONTEXT
用 Python 3.5 编写的功能替代方案。我简化了你的例子，两边只用了 5 个词。垃圾值过滤还有其他一些简化，但只需要稍作修改。我将使用 PyPI 中的包 fn 来使此功能代码更易于阅读。
from typing import List, Tuple
from itertools import groupby, filterfalse
from fn import F
首先我们需要提取列：
def getcol3(line: str) -> str:
return line.split("\t")[2]
然后我们需要将行拆分为由谓词分隔的块：
TARGET_WORDS = {"target1", "target2"}
# this is out predicate
def istarget(word: str) -> bool:
return word in TARGET_WORDS
让我们过滤垃圾并编写一个函数来获取最后和前 5 个单词：
def isjunk(word: str) -> bool:
return word == "(unknown)"
def first_and_last(words: List[str]) -> (List[str], List[str]):
first = words[:5]
last = words[-5:]
return first, last
现在，让我们获取组：
words = (F() >> (map, str.strip) >> (filter, bool) >> (map, getcol3) >> (filterfalse, isjunk))(lines)
groups = groupby(words, istarget)
现在，处理组
def is_target_group(group: Tuple[str, List[str]]) -> bool:
return istarget(group[0])
def unpack_word_group(group: Tuple[str, List[str]]) -> List[str]:
return [*group[1]]
def unpack_target_group(group: Tuple[str, List[str]]) -> List[str]:
return [group[0]]
def process_group(group: Tuple[str, List[str]]):
return (unpack_target_group(group) if is_target_group(group)
else first_and_last(unpack_word_group(group)))
最后的步骤是：
words = list(map(process_group, groups))
P.S.
这是我的测试用例：
from io import StringIO
buffer = """
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\ttarget1
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\ttarget2
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\ttarget1
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
"""
# this simulates an opened file
lines = StringIO(buffer)
给定此文件，您将获得以下输出：
[(['word', 'word', 'word', 'word', 'word'],
['word', 'word', 'word', 'word', 'word']),
(['target1'], ['target1']),
(['word', 'word', 'word', 'word'], ['word', 'word', 'word', 'word']),
(['target2'], ['target2']),
(['word', 'word', 'word', 'word', 'word'],
['word', 'word', 'word', 'word', 'word']),
(['target1'], ['target1']),
(['word', 'word', 'word', 'word'], ['word', 'word', 'word', 'word'])]
从这里您可以删除前 5 个词和最后 5 个词。
我是 Python 的新手，我目前正在使用 Python 2. 我有一些源文件，每个文件都包含大量数据（大约 1900 万行）。看起来像下面这样：
apple \t N \t apple
n&apos
garden \t N \t garden
b\ta\md
great \t Adj \t great
nice \t Adj \t (unknown)
etc
我的任务是在每个文件的第3列中搜索一些目标词，每次在语料库中找到一个目标词，就必须将这个词前后的10个词添加到一个多维词典中。
编辑： 应排除包含“&”、“\”或字符串“(unknown)”的行。
我尝试使用 readlines() 和 enumerate() 来解决这个问题，如下面的代码所示。代码做了它应该做的，但对于源文件中提供的数据量来说，它显然不够高效。
我知道 readlines() 或 read() 不应该用于巨大的数据集，因为它将整个文件加载到内存中。尽管如此，逐行读取文件，我并没有设法使用枚举方法来获取目标词前后的10个词。 我也不能使用 mmap，因为我没有在该文件上使用它的权限。
所以，我想具有一些大小限制的 readlines 方法将是最有效的解决方案。但是，为此，我会不会犯一些错误，因为每次到达大小限制的末尾时，目标词之后的 10 个词都不会被捕获，因为代码刚刚中断？
def get_target_to_dict(file):
targets_dict = {}
with open(file) as f:
for line in f:
targets_dict[line.strip()] = {}
return targets_dict
targets_dict = get_target_to_dict('targets_uniq.txt')
# browse directory and process each file
# find the target words to include the 10 words before and after to the dictionary
# exclude lines starting with <,-,; to just have raw text
def get_co_occurence(path_file_dir, targets, results):
lines = []
for file in os.listdir(path_file_dir):
if file.startswith('corpus'):
path_file = os.path.join(path_file_dir, file)
with gzip.open(path_file) as corpusfile:
# PROBLEMATIC CODE HERE
# lines = corpusfile.readlines()
for line in corpusfile:
if re.match('[A-Z]|[a-z]', line):
if '(unknown)' in line:
continue
elif '\' in line:
continue
elif '&' in line:
continue
lines.append(line)
for i, line in enumerate(lines):
line = line.strip()
if re.match('[A-Z][a-z]', line):
parts = line.split('\t')
lemma = parts[2]
if lemma in targets:
pos = parts[1]
if pos not in targets[lemma]:
targets[lemma][pos] = {}
counts = targets[lemma][pos]
context = []
# look at 10 previous lines
for j in range(max(0, i-10), i):
context.append(lines[j])
# look at the next 10 lines
for j in range(i+1, min(i+11, len(lines))):
context.append(lines[j])
# END OF PROBLEMATIC CODE
for context_line in context:
context_line = context_line.strip()
parts_context = context_line.split('\t')
context_lemma = parts_context[2]
if context_lemma not in counts:
counts[context_lemma] = {}
context_pos = parts_context[1]
if context_pos not in counts[context_lemma]:
counts[context_lemma][context_pos] = 0
counts[context_lemma][context_pos] += 1
csvwriter = csv.writer(results, delimiter='\t')
for k,v in targets.iteritems():
for k2,v2 in v.iteritems():
for k3,v3 in v2.iteritems():
for k4,v4 in v3.iteritems():
csvwriter.writerow([str(k), str(k2), str(k3), str(k4), str(v4)])
#print(str(k) + "\t" + str(k2) + "\t" + str(k3) + "\t" + str(k4) + "\t" + str(v4))
results = open('results_corpus.csv', 'wb')
word_occurrence = get_co_occurence(path_file_dir, targets_dict, results)
出于完整性原因，我复制了整个代码部分，因为它是一个函数的全部部分，该函数根据提取的所有信息创建多维字典，然后将其写入 csv 文件。
如果有任何提示或建议可以提高此代码的效率，我将不胜感激。
编辑 我更正了代码，以便它考虑到目标词前后的确切 10 个词
我的想法是在10行之前创建一个缓冲区，在10行之后创建一个缓冲区，当文件被读取时，它将被推入缓冲区之前，如果大小超过10，缓冲区将被弹出
对于后缓冲区，我从文件迭代器 1st 克隆了另一个迭代器。然后 运行 两个迭代器在循环中与克隆迭代器并行 运行 提前 10 次迭代以获得后 10 行。
这样可以避免使用 readlines() 并将整个文件加载到内存中。 希望在实际案例中对你有用
已编辑： 如果第 3 列不包含“&”、“\”、“（未知）”中的任何一个，则仅填充前后缓冲区。还将 split('\t') 更改为 split() 以便它处理所有空格或选项卡
import itertools
def get_co_occurence(path_file_dir, targets, results):
excluded_words = ['&', '\', '(unknown)'] # modify excluded words here
for file in os.listdir(path_file_dir):
if file.startswith('testset'):
path_file = os.path.join(path_file_dir, file)
with open(path_file) as corpusfile:
# CHANGED CODE HERE
before_buf = [] # buffer to store before 10 lines
after_buf = [] # buffer to store after 10 lines
corpusfile, corpusfile_clone = itertools.tee(corpusfile) # clone file iterator to access next 10 lines
for line in corpusfile:
line = line.strip()
if re.match('[A-Z]|[a-z]', line):
parts = line.split()
lemma = parts[2]
# before buffer handling, fill buffer excluded line contains any of excluded words
if not any(w in line for w in excluded_words):
before_buf.append(line) # append to before buffer
if len(before_buf)>11:
before_buf.pop(0) # keep the buffer at size 10
# next buffer handling
while len(after_buf)<=10:
try:
after = next(corpusfile_clone) # advance 1 iterator
after_lemma = ''
after_tmp = after.split()
if re.match('[A-Z]|[a-z]', after) and len(after_tmp)>2:
after_lemma = after_tmp[2]
except StopIteration:
break # copy iterator will exhaust 1st coz its 10 iteration ahead
if after_lemma and not any(w in after for w in excluded_words):
after_buf.append(after) # append to buffer
# print 'after',z,after, ' - ',after_lemma
if (after_buf and line in after_buf[0]):
after_buf.pop(0) # pop off one ready for next
if lemma in targets:
pos = parts[1]
if pos not in targets[lemma]:
targets[lemma][pos] = {}
counts = targets[lemma][pos]
# context = []
# look at 10 previous lines
context= before_buf[:-1] # minus out current line
# look at the next 10 lines
context.extend(after_buf)
# END OF CHANGED CODE
# CONTINUE YOUR STUFF HERE WITH CONTEXT
用 Python 3.5 编写的功能替代方案。我简化了你的例子，两边只用了 5 个词。垃圾值过滤还有其他一些简化，但只需要稍作修改。我将使用 PyPI 中的包 fn 来使此功能代码更易于阅读。
from typing import List, Tuple
from itertools import groupby, filterfalse
from fn import F
首先我们需要提取列：
def getcol3(line: str) -> str:
return line.split("\t")[2]
然后我们需要将行拆分为由谓词分隔的块：
TARGET_WORDS = {"target1", "target2"}
# this is out predicate
def istarget(word: str) -> bool:
return word in TARGET_WORDS
让我们过滤垃圾并编写一个函数来获取最后和前 5 个单词：
def isjunk(word: str) -> bool:
return word == "(unknown)"
def first_and_last(words: List[str]) -> (List[str], List[str]):
first = words[:5]
last = words[-5:]
return first, last
现在，让我们获取组：
words = (F() >> (map, str.strip) >> (filter, bool) >> (map, getcol3) >> (filterfalse, isjunk))(lines)
groups = groupby(words, istarget)
现在，处理组
def is_target_group(group: Tuple[str, List[str]]) -> bool:
return istarget(group[0])
def unpack_word_group(group: Tuple[str, List[str]]) -> List[str]:
return [*group[1]]
def unpack_target_group(group: Tuple[str, List[str]]) -> List[str]:
return [group[0]]
def process_group(group: Tuple[str, List[str]]):
return (unpack_target_group(group) if is_target_group(group)
else first_and_last(unpack_word_group(group)))
最后的步骤是：
words = list(map(process_group, groups))
P.S.
这是我的测试用例：
from io import StringIO
buffer = """
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\ttarget1
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\ttarget2
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\ttarget1
_\t_\tword
_\t_\t(unknown)
_\t_\tword
_\t_\tword
_\t_\tword
"""
# this simulates an opened file
lines = StringIO(buffer)
给定此文件，您将获得以下输出：
[(['word', 'word', 'word', 'word', 'word'],
['word', 'word', 'word', 'word', 'word']),
(['target1'], ['target1']),
(['word', 'word', 'word', 'word'], ['word', 'word', 'word', 'word']),
(['target2'], ['target2']),
(['word', 'word', 'word', 'word', 'word'],
['word', 'word', 'word', 'word', 'word']),
(['target1'], ['target1']),
(['word', 'word', 'word', 'word'], ['word', 'word', 'word', 'word'])]
从这里您可以删除前 5 个词和最后 5 个词。