Python чтение из файла и запись: 'charmap' codec can't encode character

Question

Имеется текстовый файл овер 20гб В нем необходимо изменить кодировку, заменить разделитель для дальнейшей прогрузки в бд. По ходу конвертации появляются ошибки, незнаю че с ними делать, коды символов разные:

#Error: 'charmap' codec can't encode character '\xb9' in position 67: `character` maps to <undefined>, in line 0 str =
#Error: 'charmap' codec can't encode character '\x98' in position 197: character maps to <undefined>, in line 0 str =

И второй вопрос, можно ли как-то распараллелить процесс? может писать в разные файлы? В один поток за рабочий день файл не конвертируется. Но тогда вопрос, как делить исходный файл, читаю про потоки, но пока не пробовал. Исходный скрипт:

import codecs
import time
import re
file = codecs.open('\\\\tsclient\\C\\Users\\User\\Documents\\MDM\\MDM\\H_IDENTIFIER_000000.txt','r', 'utf-8') #добавил
out = open('\\\\tsclient\\C\\Users\\User\\Documents\\MDM\\MDM\\H_IDENTIFIER.txt', 'w')
n = 0
print('Start :'.format(), time.strftime('%c'))
for i in file:
    try:       
        out.write(re.sub('\s+','',i.replace('  ',';')) + '\n')       
    except Exception as e:
        print('Error: {}, in line {} str = '.format(e, n, i))
        pass
out.close()
print('end :'.format(), time.strftime('%c'))

Answer 1

Попробуйте такой код:

import re
import time
import codecs
import multiprocessing as mt
from queue import Empty as queue_Empty

INP_FILENAME = '\\\\tsclient\\C\\Users\\User\\Documents\\MDM\\MDM\\H_IDENTIFIER_000000.txt'
OUT_FILENAME = '\\\\tsclient\\C\\Users\\User\\Documents\\MDM\\MDM\\H_IDENTIFIER.txt'
ERR_FILENAME = '\\\\tsclient\\C\\Users\\User\\Documents\\MDM\\MDM\\H_IDENTIFIER_err.txt'

NUM_THREAD = `8


def read_line(filename, inp_queue):
    with codecs.open(INP_FILENAME,'r', 'utf-8') as f:
        for idx, line in enumerate(f): 
            #DEBUG print((idx, line))
            inp_queue.put((idx, line))
    pass


def processed_line(inp_queue, out_queue):
    while True:
        try:
            idx, line = inp_queue.get(): 
            #DEBUG print((idx, line))
            out_queue.put(re.sub(r'\s+','', line.replace('  ',';')) + '\n')
        except Exception:
            out_queue.put((idx, line))
    pass 


def write_line(out_filename, out_queue, err_filename):
    err_f = open(err_filename, 'w')
    with open(out_filename, 'w') as f:
        while True:
            data = out_queue.get(timeout=10): 
            #DEBUG print(data)
            try:
                f.write(data)
            except Exception:
                if isinstance(data, tuple):
                    err_f.write(str(data))
                else:
                    err_f.write(str(data[0]))
    pass


def main():
    inp_queue = mt.Queue(10000)
    out_queue = mt.Queue(10000)

    lprocess = []
    lprocess.append(mt.Process(target=read_line, args=(INP_FILENAME, inp_queue), daemon=True))
    for _ in range(NUM_THREAD):
        lprocess.append(mt.Process(target=processed_line, args=(inp_queue, out_queue), daemon=True))
    lprocess.append(mt.Process(target=write_line, args=(OUT_FILENAME, out_queue, ERR_FILENAME), daemon=True))

    for process in lprocess:
        process.start()

    lprocess.pop(0).join()
    print('end read')
    lprocess.pop(-1).join()
    print('end :'.format(), time.strftime('%c'))

if __name__ == '__main__':
    main()

P.S. Сюдя по путям вы читаете файл которых находиться на другом компютере, а это может быть очень медленно. Лучше перенесите место запуска или файл.

Answer 2

Ваша функция - worker(). Распаралелить можно, читая часть файла, и записывая обработанный результат в разные файлы

subprocess_run : запускать скрипт 'sub.py' в новой python сесии, те у кажного процесса собственный GIL, по идее самый быстрый вариант

concurrent_run, multiprocessing_run : чтение файла по частям, чем больше lines_limit, тем больше процесс за один подход обработает данных, тем быстрее общее время

single_run: чтение файла целиком, однопоточное выполнение (для сравнения)

import re
import subprocess
import multiprocessing
import concurrent.futures
from itertools import islice
import timeit


def worker(args):
    """
    читать rfile, начиная со строки begin_line до строки end_line
    результат писать в wfile(писать в разные файлы)
    """
    rfile, wfile, begin_line, end_line, enc = args
    with open(rfile, encoding=enc, errors='ignore') as rf, open(wfile, 'w', encoding=enc, errors='ignore') as wf:
        for line in islice(rf, begin_line, end_line):
            wf.write('%s\n' % re.sub('\s+', '', line.replace('  ', ';')))


def subprocess_run(file='sub.py'):
    """выполнить (2 процесса), через новую python-сессию"""
    m = max_line//2    # обработать строк за проход
    with subprocess.Popen(['python', file, str([file_read, file_write % 1, 0, m, enc])]), \
         subprocess.Popen(['python', file, str([file_read, file_write % 2, m, max_line, enc])]):
        pass

def multiprocessing_run():
    """выполнить через multiprocessing"""
    with multiprocessing.Pool(processes=2) as pool:
        pool.map(worker, worker_args)

def concurrent_run():
    """выполнить через ProcessPoolExecutor"""
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as pool:
        pool.map(worker, worker_args)

def single_run():
    """выполнить однопоточно"""
    worker([file_read, file_write % '_', 0, max_line, enc])


if __name__ == '__main__':
    # исходные данные
    enc = 'utf-8'
    file_read = 'test2.txt'
    file_write = 'test2_out_%s.txt'
    start_line, stop_line = 0, 0
    lines_limit = 1500000  # обработать строк за проход
    max_line = len(list(open(file_read)))  # кол-во строк файла

    # для multiprocessing_run, concurrent_run
    worker_args = []
    while start_line < max_line:
        start_line = stop_line
        stop_line += lines_limit
        args = file_read, file_write % start_line, start_line, stop_line, enc
        worker_args.append(args)

    # выполнить и замерить производительность
    for fn in [subprocess_run, single_run, concurrent_run, multiprocessing_run]:
        print(timeit.Timer(fn).repeat(1, 1), 'сек', fn.__name__)

файл 'sub.py'

import sys
import re
from itertools import islice

rfile, wfile, begin_line, end_line, enc = eval(sys.argv[1])

with open(rfile, encoding=enc, errors='ignore') as rf, open(wfile, 'w', encoding=enc, errors='ignore') as wf:
    for line in islice(rf, begin_line, end_line):
        wf.write('%s\n' % re.sub('\s+', '', line.replace('  ', ';')))

быстродействие, исходный файл 'test2.txt' - 120Mb ~ 3000000 строк :

[16.207453404993398] сек subprocess_run
[29.336879417459787] сек single_run
[18.94762167044125] сек concurrent_run
[20.828409269737875] сек multiprocessing_run