[polars] 데이터 EDA 코드 만들기 (컬럼수, 행수, 결측값 등)

Polars 라이브러리를 사용하여 데이터 탐색적 분석(EDA) 코드를 만드는 방법을 소개합니다. 특히, 컬럼 수, 행 수, 결측값 수 등 기본적인 EDA 정보를 계산하는 코드를 다룹니다.

 

import chardet
#polars
import polars as pl

file_name = "데이터 경로"
charenc = detect_enc(file_name)
# open file
data = pl.read_csv(file_name, encoding = charenc, ignore_errors= True,  low_memory=False, truncate_ragged_lines = True, infer_schema_length=0)
# 컬럼명
column_list = list(data.columns)
# 컬럼 수
column_count = data.shape[1]
# 행 수
row_count = data.shape[0]
# 결측값 수
# 정확한 계산을 위해 특수문자와 공백값을 전처리함
data = data.with_columns(
pl.when(pl.col(pl.Utf8).is_in(["", " ", "nan", "null", "na"]))
.then(None)
.otherwise(pl.col(pl.Utf8))  # keep original value
.name.keep()
)

pattern = "[-?@#$%^&*\(\)]"
for col in data.columns:
    data = data.with_columns(pl.col(col).str.replace_all(pattern, ""))

null_count = pl.Series([value[0] for value in data.null_count().head(1)]).to_list()

# 데이터 프레임으로 저장
list_of_data = [file_name, file_format, charenc, column_count, row_count, column_list, null_count]
tmp_df = pl.DataFrame( {
'파일명' : [list_of_data[0]],
'확장자' : [list_of_data[1]],
'인코딩' : [list_of_data[2]],
'컬럼수' : [list_of_data[3]],
'행수' : [list_of_data[4]],
'컬럼명' : [list_of_data[5]],
'결측값수' : [list_of_data[6]]
})

 

결과 예시

 

 

1. 데이터 불러오기 및 인코딩 설정

detect_enc 함수를 사용하여 파일의 인코딩을 자동으로 감지합니다. # pip install chardet

pl.read_csv 함수를 사용하여 데이터를 불러옵니다.

2. 컬럼 및 행 정보

list(data.columns)를 사용하여 컬럼 리스트를 추출
data.shape[1]을 사용하여 컬럼 수를 계산함
data.shape[0]을 사용하여 행 수를 계산함

3. 결측값 처리

pl.when 함수를 사용하여 특수문자와 공백값을 None으로 변환

 

- 문자열 정제

정규표현식을 사용하여 특수문자를 정제함

 

 

ps. pandas로 같은 코드를 작성해서 실행했을 땐, 거의 2-3시간 걸리던 게 polars로 바꾸니 3분만에 끝남

 

 

 

특정 폴더의 모든 파일에 적용하는 경우

 

import polars as pl
import numpy as np
import pandas as pd
import time
from tqdm import tqdm
import os
import chardet


folder = r'여러 파일이 존재하는 경로'
# csv 또는 xlsx 파일만 리스트로 가져옵니다


# csv 파일의 encoding 찾기
def detect_enc(file_name) :
    f = open(folder + '/' + file_name, 'rb')
    rawdata = f.readlines()[0]
    rawdata_enc = chardet.detect(rawdata)
    charenc = rawdata_enc['encoding']
    f.close()
    
    return charenc

list_file = [file for file in os.listdir(folder) if file.endswith('.csv') or file.endswith('.xlsx')]
print(f'{folder} 시작합니다 -------------------------------------------------------------------------------------------------------')

# 데이터 틀 만들기
df = pl.DataFrame({'파일명' : None, '확장자' : None, '인코딩' : None, '컬럼수' : None, '행수' : None, '컬럼명' : None, '결측값수' : None})
df = df.drop_nulls()
# str	str	str	i64	i64	list[str]	list[i64]
df = df.cast({"파일명": pl.Utf8, "확장자": pl.Utf8, "인코딩": pl.Utf8, "컬럼수": pl.Int64, "행수": pl.Int64, "컬럼명": pl.List(pl.Utf8), "결측값수": pl.List(pl.Int64)})

pattern = "[-?@#$%^&*\(\)]"

for file_name in tqdm(list_file, total=len(list_file)):
    #file_id = file_name.split('_')[0]
    file_format = str(file_name.split('.')[-1]).upper()
    try : 
        #     # xlsx
        if file_format == 'XLSX':
            #print(file_name, file_format)
            charenc = ''
            # open file
            data = pl.read_excel(folder + '/' + file_name, read_options={"infer_schema_length": 1000000})
            # 컬럼명
            column_list = list(data.columns)
            # 컬럼 수
            column_count = data.shape[1]
            # 행 수
            row_count = data.shape[0]
            # 결측값 수
            data = data.with_columns(
            pl.when(pl.col(pl.Utf8).is_in(["", " ", "nan", "null", "na"]))
            .then(None)
            .otherwise(pl.col(pl.Utf8))  # keep original value
            .name.keep()
            )
            for col in data.columns:
                data = data.with_columns(pl.col(col).str.replace_all(pattern, ""))

            null_count = pl.Series([value[0] for value in data.null_count().head(1)]).to_list()
            
            # row
            list_of_data = [file_name,file_format, charenc, column_count, row_count, column_list, null_count] 
                
        # csv, xlsx 아닌 경우 pass 
        elif file_format == 'CSV':
            #print(file_name, file_format)
            try:    
                charenc = detect_enc(file_name)

                # open file
                data = pl.read_csv(folder + '/' + file_name, encoding = charenc, ignore_errors= True,  low_memory=False, truncate_ragged_lines = True, infer_schema_length=0) # dtype = str
                # 컬럼명
                column_list = list(data.columns)
                # 컬럼 수
                column_count = data.shape[1]
                # 행 수
                row_count = data.shape[0]
                # 결측값 수
                data = data.with_columns(
                pl.when(pl.col(pl.Utf8).is_in(["", " ", "nan", "null", "na"]))
                .then(None)
                .otherwise(pl.col(pl.Utf8))  # keep original value
                .name.keep()
                )
                for col in data.columns:
                    data = data.with_columns(pl.col(col).str.replace_all(pattern, ""))

                null_count = pl.Series([value[0] for value in data.null_count().head(1)]).to_list()
                
                # 저장
                list_of_data = [file_name, file_format, charenc, column_count, row_count, column_list, null_count]
                    
            except:
                charenc = 'cp949'
                
                # open file
                data = pl.read_csv(folder + '/' + file_name, encoding = charenc, ignore_errors= True,  low_memory=False, truncate_ragged_lines = True, infer_schema_length=0)
                # 컬럼명
                column_list = list(data.columns)
                # 컬럼 수
                # 컬럼 수
                column_count = data.shape[1]
                # 행 수
                row_count = data.shape[0]
                # 결측값 수
                data = data.with_columns(
                pl.when(pl.col(pl.Utf8).is_in(["", " ", "nan", "null", "na"]))
                .then(None)
                .otherwise(pl.col(pl.Utf8))  # keep original value
                .name.keep()
                )
                for col in data.columns:
                    data = data.with_columns(pl.col(col).str.replace_all(pattern, ""))
                
                null_count = pl.Series([value[0] for value in data.null_count().head(1)]).to_list()
                # 저장
                list_of_data = [file_name, file_format, charenc, column_count, row_count, column_list, null_count]
        else:
            pass
        
    except : 
        print('open_error')
        list_of_data = [file_name,file_format, 'open_error', None, None, None, None]
        
    tmp_df = pl.DataFrame( {
            '파일명' : [list_of_data[0]],
            '확장자' : [list_of_data[1]],
            '인코딩' : [list_of_data[2]],
            '컬럼수' : [list_of_data[3]],
            '행수' : [list_of_data[4]],
            '컬럼명' : [list_of_data[5]],
            '결측값수' : [list_of_data[6]]
            })
    tmp_df = tmp_df.cast({"파일명": pl.Utf8, "확장자": pl.Utf8, "인코딩": pl.Utf8, "컬럼수": pl.Int64, "행수": pl.Int64, "컬럼명": pl.List(pl.Utf8), "결측값수": pl.List(pl.Int64)})

    df = df.vstack(tmp_df)