我有100个大文件,每个文件大约5 5GB。我需要根据它们的内容将它们拆分成文件。大文件有很多行,每行都是这样的
{"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 is数据需要2小时。
那么有没有更好的方法来处理数据呢?
非常感谢你的帮助。
发布于 2018-12-18 20:57:03
我推测主要耗费时间的过程是文件IO操作。你能分解一下运行时间并检查一下吗?
另一个原因可能是JSON解析器。有关更多信息,请查看this线程。
发布于 2018-12-18 21:34:57
你能把这些文件分类吗?如果是,请尝试不要将每一行都解析为JSON,只解析具有新ID的行。
像这样的东西?
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')https://stackoverflow.com/questions/53833336
复制相似问题