Есть несколько вопросов SO, посвященных той или иной форме этой темы, но все они кажутся ужасно неэффективными для удаления только одной строки из файла csv (обычно это связано с копированием всего файла). Если у меня есть csv, отформатированный так:
Каков наиболее эффективный способ удаления строки Сара? Если возможно, я бы хотел избежать копирования всего файла.
Ответ 3
Редактирование файлов на месте - это задача, пронизанная gotchas (так же, как изменение итерации при итерации по ней) и обычно не стоит проблем. В большинстве случаев запись во временный файл (или рабочую память, в зависимости от того, что у вас больше - пространство для хранения или оперативная память), то удаление исходного файла и замена исходного файла на временный файл будут одинаково эффективными, как попытка выполнить такой же предмет на месте.
Но, если вы настаиваете, вот обобщенное решение:
import os
def remove_line(path, comp):
with open(path, "r+b") as f: # open the file in rw mode
mod_lines = 0 # hold the overwrite offset
while True:
last_pos = f.tell() # keep the last line position
line = f.readline() # read the next line
if not line: # EOF
break
if mod_lines: # we've already encountered what we search for
f.seek(last_pos - mod_lines) # move back to the beginning of the gap
f.write(line) # fill the gap with the current line
f.seek(mod_lines, os.SEEK_CUR) # move forward til the next line start
elif comp(line): # search for our data
mod_lines = len(line) # store the offset when found to create a gap
f.seek(last_pos - mod_lines) # seek back the extra removed characters
f.truncate() # truncate the rest
Это приведет к удалению только строки, соответствующей предоставленной функции сравнения, а затем перебирает оставшуюся часть файла, перемещая данные по строке "удаленный". Вам не нужно будет загружать остальную часть файла в вашу рабочую память. Чтобы проверить его, test.csv содержит:
fname,lname,age,sex
John,Doe,28,m
Sarah,Smith,27,f
Xavier,Moore,19,m
Вы можете запустить его как:
remove_line("test.csv", lambda x: x.startswith(b"Sarah"))
И вы получите test.csv с удаленной линией Sarah:
fname,lname,age,sex
John,Doe,28,m
Xavier,Moore,19,m
Имейте в виду, что мы передаем функцию сравнения bytes как файл открывается в двоичном режиме, чтобы поддерживать согласованные разрывы строк при усечении/перезаписи.
ОБНОВЛЕНИЕ: меня интересовала реальная производительность различных приемов, представленных здесь, но у меня не было времени, чтобы проверить их вчера, поэтому, немного задержавшись, я создал тест, который мог бы пролить свет на него. Если вас интересуют только результаты, прокрутите весь путь вниз. Сначала я объясню, что я сравнивал и как я настроил тест. Я также предоставил все сценарии, чтобы вы могли запускать те же тесты в своей системе.
Что касается того, что, я протестировал все упомянутые методы в этом и других ответах, а именно, замену строки с использованием временного файла (функции temp_file_*) и использование функций редактирования на месте (in_place_*). У меня есть оба из тех, которые настроены в потоковой передаче (чтение строки за строкой, *_stream функции) и памяти (чтение остальной части файла в рабочей памяти, *_wm функции). Я также добавил метод удаления строки на месте с помощью модуля mmap (функция in_place_mmap). Контрольный сценарий, содержащий все функции, а также небольшую логику, управляемую через CLI, выглядит следующим образом:
#!/usr/bin/env python
import mmap
import os
import shutil
import sys
import time
def get_temporary_path(path): # use tempfile facilities in production
folder, filename = os.path.split(path)
return os.path.join(folder, "~$" + filename)
def temp_file_wm(path, comp):
path_out = get_temporary_path(path)
with open(path, "rb") as f_in, open(path_out, "wb") as f_out:
while True:
line = f_in.readline()
if not line:
break
if comp(line):
f_out.write(f_in.read())
break
else:
f_out.write(line)
f_out.flush()
os.fsync(f_out.fileno())
shutil.move(path_out, path)
def temp_file_stream(path, comp):
path_out = get_temporary_path(path)
not_found = True # a flag to stop comparison after the first match, for fairness
with open(path, "rb") as f_in, open(path_out, "wb") as f_out:
while True:
line = f_in.readline()
if not line:
break
if not_found and comp(line):
continue
f_out.write(line)
f_out.flush()
os.fsync(f_out.fileno())
shutil.move(path_out, path)
def in_place_wm(path, comp):
with open(path, "r+b") as f:
while True:
last_pos = f.tell()
line = f.readline()
if not line:
break
if comp(line):
rest = f.read()
f.seek(last_pos)
f.write(rest)
break
f.truncate()
f.flush()
os.fsync(f.fileno())
def in_place_stream(path, comp):
with open(path, "r+b") as f:
mod_lines = 0
while True:
last_pos = f.tell()
line = f.readline()
if not line:
break
if mod_lines:
f.seek(last_pos - mod_lines)
f.write(line)
f.seek(mod_lines, os.SEEK_CUR)
elif comp(line):
mod_lines = len(line)
f.seek(last_pos - mod_lines)
f.truncate()
f.flush()
os.fsync(f.fileno())
def in_place_mmap(path, comp):
with open(path, "r+b") as f:
stream = mmap.mmap(f.fileno(), 0)
total_size = len(stream)
while True:
last_pos = stream.tell()
line = stream.readline()
if not line:
break
if comp(line):
current_pos = stream.tell()
stream.move(last_pos, current_pos, total_size - current_pos)
total_size -= len(line)
break
stream.flush()
stream.close()
f.truncate(total_size)
f.flush()
os.fsync(f.fileno())
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Usage: {} target_file.ext <search_string> [function_name]".format(__file__))
exit(1)
target_file = sys.argv[1]
search_func = globals().get(sys.argv[3] if len(sys.argv) > 3 else None, in_place_wm)
start_time = time.time()
search_func(target_file, lambda x: x.startswith(sys.argv[2].encode("utf-8")))
# some info for the test runner...
print("python_version: " + sys.version.split()[0])
print("python_time: {:.2f}".format(time.time() - start_time))
Следующим шагом будет создание тестера, который будет выполнять эти функции как можно более изолированную среду, пытаясь получить справедливый ориентир для каждого из них. Мой тест структурирован как:
- Три примера данных CSV генерируются как матрицы 1Mx10 (файлы ~ 200 МБ) случайных чисел с идентифицируемой линией, помещенной в начале, середине и конце их соответственно, тем самым генерируя тестовые примеры для трех экстремальных сценариев.
- Файлы основных образцов данных копируются как временные файлы (поскольку удаление строк является разрушительным) перед каждым тестом.
- Различные способы синхронизации файлов и очистки кеша используются для обеспечения чистых буферов перед началом каждого теста.
- Тесты выполняются с использованием наивысшего приоритета (
chrt -f 99) через /usr/bin/time для теста, поскольку Python не может действительно быть уверенным в точном измерении его производительности в таких сценариях. - Для сглаживания непредсказуемых колебаний выполняются по меньшей мере три прогона каждого теста.
- Тесты также выполняются в Python 2.7 и Python 3.6 (CPython), чтобы убедиться, что между версиями существует согласованность производительности.
- Все контрольные данные собираются и сохраняются как CSV для будущего анализа.
К сожалению, у меня не было системы, где я мог бы выполнить тест полностью изолированным, чтобы мои числа были получены от запуска его в гипервизоре. Это означает, что производительность ввода-вывода, вероятно, очень искажена, но она также должна влиять на все тесты, которые по-прежнему предоставляют сопоставимые данные. В любом случае, вы можете запустить этот тест в своей собственной системе, чтобы получить результаты, с которыми вы можете иметь отношение.
Я установил тестовый скрипт, выполняющий вышеупомянутый сценарий, как:
#!/usr/bin/env python
import collections
import os
import random
import shutil
import subprocess
import sys
import time
try:
range = xrange # cover Python 2.x
except NameError:
pass
try:
DEV_NULL = subprocess.DEVNULL
except AttributeError:
DEV_NULL = open(os.devnull, "wb") # cover Python 2.x
SAMPLE_ROWS = 10**6 # 1M lines
TEST_LOOPS = 3
CALL_SCRIPT = os.path.join(os.getcwd(), "remove_line.py") # the above script
def get_temporary_path(path):
folder, filename = os.path.split(path)
return os.path.join(folder, "~$" + filename)
def generate_samples(path, data="LINE", rows=10**6, columns=10): # 1Mx10 default matrix
sample_beginning = os.path.join(path, "sample_beg.csv")
sample_middle = os.path.join(path, "sample_mid.csv")
sample_end = os.path.join(path, "sample_end.csv")
separator = os.linesep
middle_row = rows // 2
with open(sample_beginning, "w") as f_b, \
open(sample_middle, "w") as f_m, \
open(sample_end, "w") as f_e:
f_b.write(data)
f_b.write(separator)
for i in range(rows):
if not i % middle_row:
f_m.write(data)
f_m.write(separator)
for t in (f_b, f_m, f_e):
t.write(",".join((str(random.random()) for _ in range(columns))))
t.write(separator)
f_e.write(data)
f_e.write(separator)
return ("beginning", sample_beginning), ("middle", sample_middle), ("end", sample_end)
def normalize_field(field):
field = field.lower()
while True:
s_index = field.find('(')
e_index = field.find(')')
if s_index == -1 or e_index == -1:
break
field = field[:s_index] + field[e_index + 1:]
return "_".join(field.split())
def encode_csv_field(field):
if isinstance(field, (int, float)):
field = str(field)
escape = False
if '"' in field:
escape = True
field = field.replace('"', '""')
elif "," in field or "\n" in field:
escape = True
if escape:
return ('"' + field + '"').encode("utf-8")
return field.encode("utf-8")
if __name__ == "__main__":
print("Generating sample data...")
start_time = time.time()
samples = generate_samples(os.getcwd(), "REMOVE THIS LINE", SAMPLE_ROWS)
print("Done, generation took: {:2} seconds.".format(time.time() - start_time))
print("Beginning tests...")
search_string = "REMOVE"
header = None
results = []
for f in ("temp_file_stream", "temp_file_wm",
"in_place_stream", "in_place_wm", "in_place_mmap"):
for s, path in samples:
for test in range(TEST_LOOPS):
result = collections.OrderedDict((("function", f), ("sample", s),
("test", test)))
print("Running {function} test, {sample} #{test}...".format(**result))
temp_sample = get_temporary_path(path)
shutil.copy(path, temp_sample)
print(" Clearing caches...")
subprocess.call(["sudo", "/usr/bin/sync"], stdout=DEV_NULL)
with open("/proc/sys/vm/drop_caches", "w") as dc:
dc.write("3\n") # free pagecache, inodes, dentries...
# you can add more cache clearing/invalidating calls here...
print(" Removing a line starting with '{}'...".format(search_string))
out = subprocess.check_output(["sudo", "chrt", "-f", "99",
"/usr/bin/time", "--verbose",
sys.executable, CALL_SCRIPT, temp_sample,
search_string, f], stderr=subprocess.STDOUT)
print(" Cleaning up...")
os.remove(temp_sample)
for line in out.decode("utf-8").split("\n"):
pair = line.strip().rsplit(": ", 1)
if len(pair) >= 2:
result[normalize_field(pair[0].strip())] = pair[1].strip()
results.append(result)
if not header: # store the header for later reference
header = result.keys()
print("Cleaning up sample data...")
for s, path in samples:
os.remove(path)
output_file = sys.argv[1] if len(sys.argv) > 1 else "results.csv"
output_results = os.path.join(os.getcwd(), output_file)
print("All tests completed, writing results to: " + output_results)
with open(output_results, "wb") as f:
f.write(b",".join(encode_csv_field(k) for k in header) + b"\n")
for result in results:
f.write(b",".join(encode_csv_field(v) for v in result.values()) + b"\n")
print("All done.")
Наконец (и TL; DR): вот мои результаты. Я извлекаю только лучшие данные времени и памяти из набора результатов, но здесь вы можете получить полные результирующие множества: Python 2.7 Raw Test Data и Python 3.6 Raw Test Data.
![Python File Line Removal - Selected Results]()
Основываясь на данных, которые я собрал, несколько заключительных заметок:
- Если рабочая память является проблемой (работа с исключительно большими файлами и т.д.), Только функции
*_stream обеспечивают небольшую площадь. На Python 3.xa на середине будет метод mmap. - Если проблема с хранилищем является проблемой, только функции
in_place_* являются жизнеспособными. - Если оба они недостаточны, единственной последовательной методикой является
in_place_stream но за счет времени обработки и увеличенных вызовов ввода-вывода (по сравнению с функциями *_wm). -
in_place_* опасны, поскольку они могут привести к повреждению данных, если они остановлены на середине. temp_file_* (без проверки целостности) опасны только для файловых систем без транзакций.
