Как разбить большой файл (5 ГБ) на маленькие файлы в соответствии с его содержимым?

Question

У меня есть 100 больших файлов, и каждый составляет около 5 ГБ.Мне нужно разбить их на файлы на основе его содержимого.Большие файлы имеют много строк, каждая строка выглядит следующим образом:

{"task_op_id": 143677789, "task_op_time": 1530927931, "task_op_tag": 1, "create_time": 1530923701, "status": 2}

, и мне нужно разделить содержимое на основе task_op_id, каждый большой файл имеет 350 разных task_op_id, поэтому каждый должен генерировать 350 разных маленьких файлов,каждый из них имеет одинаковое содержимое task_op_id.

Мой проверенный метод:

def split_to_id_file(original_file):
    destination_file = 'processed_data2/data_over_one_id/break_into_ids/'
    with open(original_file) as f1:
        for line in f1:
            data_dict = json.loads(line)
            task_op_id = data_dict['task_op_id']
            with open(destination_file+str(task_op_id), 'a+') as f2:
                json.dump(data_dict, f2, ensure_ascii=False)
                f2.write('\n')
# multiprocessing with pool
def multiprocessing_pool(workers_number, job, files_list):
    p = Pool(workers_number)
    p.map(job, files_list)


def main():
    input_path = 'processed_data2/data_over_one_id'
    files_list = [join(input_path, f) for f in listdir(input_path)
              if isfile(join(input_path, f))
              and join(input_path, f).split('/')[-1].startswith('uegaudit')]
    multiprocessing_pool(80, split_to_id_file, files_list)


if __name__ == '__main__':
    main()

Но скорость слишком низкая, обработка данных 10 ГБ требует 2 часа.

Так что естьлучший способ обработки данных?

Большое спасибо за помощь.

Answer 1

Можете ли вы отсортировать эти файлы?Если да, попробуйте не анализировать каждую строку как JSON, только эти с новым идентификатором.

Как-то так?

def get_id(json_line): 
  data_dict = json.loads(json_line)
  return data_dict['task_op_id']

def split_to_id_file(original_file):
  current_id = 'blabla_xxxxxxxx'
  destination_file = 'processed_data2/data_over_one_id/break_into_ids/'
  with open(original_file) as f1:
    for line in f1:
        if current_id not in line:
          if not f2.closed:
            f2.close()
          task_op_id = get_id(line)
          current_id = "\"task_op_id\": " + task_op_id
          f2 = open(destination_file+str(task_op_id), 'a+')
        f2.write(line+'\n')

Answer 2

Я полагаю, что основной процесс получения времени - это операции ввода-вывода файла.Можете ли вы сократить время выполнения и проверить это?

Другой причиной может быть анализатор JSON.Проверьте эту тему для получения дополнительной информации.