[Langchain] Dask로 대용량 판례 데이터 처리하기

판례 json 데이터가 5만 여개가 넘고, 각 데이터를 하나의 CSV로 처리하여 관리하고 싶었다. 

나중에는 db를 활용해야겠지만, 현재는 판례 데이터를 병합하여 하나의 데이터프레임으로 만들어 로컬에서 관리할 것이다. 

 

이를 위해 json을 읽고 데이터프레임으로 빠르게 변환하는 방법인 dask를 사용하였다. 

import dask.bag as db
import json
import glob

directory_path = "./1.판례"

json_files = glob.glob(f"{directory_path}/*.json")

bag = db.from_sequence(json_files)

def load_json(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        return json.load(file)

def handle_missing_values(record):
    for key, value in record.items():
        if key == '판례일련번호' or key == '사건종류코드':
            if value is None or value == '':
                record[key] = 0
        elif value is None or value == '':
            record[key] = ""
    return record

    return record
data_bag = bag.map(load_json).map(handle_missing_values)

df = data_bag.to_dataframe()

print(df.compute())

 

간단한 코드만으로 10여분 걸리던 변환 작업이 47초만에 완료됐다...ㄷㄷ

 

단 dask dataframe에서는 자유자재로 데이터를 다루기에는 지연이 많이 발생한다.

따라서 일단 처리된 데이터를 csv로 저장하고,

전처리 작업에서는  pandas dataframe에서 다시 로드하여 사용해야 겠다. 

 

결과 또한 pandas가 깔끔하게 보여서 좋다!