Python reading from file and writing: 'charmap' codec can't encode character

Your function is worker (). You can parallelize by reading a part of a file and writing the processed result into different files.

subprocess_run: run the script 'sub.py' in the new python session, those in each process have their own GIL, in theory the fastest option

concurrent_run, multiprocessing_run: reading a file in parts, the more lines_limit, the more process will process data in one approach, the faster the total time

single_run: read the entire file, single-threaded execution (for comparison)

 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' file

 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(' ', ';'))) 

speed, the source file 'test2.txt' - 120Mb ~ 3000000 lines:

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

Try this code:

 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() 

PS Going through the paths you read the file which is located on another computer, and this can be very slow. It is better to transfer the place of launch or file.