대량의 데이터 다루기, 누락된 데이터 처리, 클래스 활용

 

 

• 어제 같이 국비지원 학원을 다녔던 친구를 만나서 가볍게 한잔하고 왔더니 한시간 더 자버렸다ㅋ
• 그래도 적당히 마셔서 다행이다 휘유~
• 그리고 돈 좀 아껴쓰자!!!

 

 

 

 

---

 

 

 

 

 

1. 자료구조를 이용해 수만 개의 데이터 처리하기

• 어떤 자료 구조를 이용해야 수만 개의 데이터를 빠르게 다룰 수 있을까?
• 파이썬으로 대량의 데이터를 만들어보고 직접해보자
• 성(last name)과 이름(first name)을 30개씩 준비한다
• 가능한 조합은 30 * 30 = 900개이다

import random
import numpy as np
import pandas as pd

import heapq
import time

last_names = [
    "Smith", "Johnson", "Williams", "Jones", "Brown",
    "Davis", "Miller", "Wilson", "Moore", "Taylor",
    "Anderson", "Thomas", "Jackson", "White", "Harris",
    "Martin", "Thompson", "Garcia", "Martinez", "Robinson",
    "Clark", "Rodriguez", "Lewis", "Lee", "Walker",
    "Hall", "Allen", "Young", "Hernandez", "King"
]

first_names = [
    "James", "Mary", "Robert", "Patricia"	, "John",
    "Jennifer", "Michael", "Linda", "David", "Elizabeth",
    "William", "Barbara", "Richard", "Susan", "Joseph",
    "Jessica", "Thomas", "Sarah", "Charles", "Karen",
    "Christopher", "Lisa", "Daniel", "Nancy", "Matthew",
    "Betty", "Anthony", "Margaret", "Mark", "Sandra"
]

def generate_name():
    # 랜덤으로 하나의 성(last name) 추출
    last_name = random.choice(last_names)
    # 랜덤으로 하나의 이름(first name) 추출
    first_name = random.choice(first_names)
    return first_name + " " + last_name

• 5개의 학과를 준비한다
• 성적은 평균(mean)이 50점, 표준편차가 10점인 정규 분포를 따른다고 가정한다
• 3개의 학년을 준비한다

departments = [
    "Computer Science",
    "Mechanical Engineering",
    "Biomedical Engineering",
    "Radiology",
    "Psychology"
]

def genearte_department():
    # 랜덤으로 하나의 학과(department) 추출
    return random.choice(departments)

mu = 50
sigma = 10

def generate_score():
    # 랜덤으로 하나의 성적(score) 추출
    return np.random.normal(mu, sigma, 1)[0]

grades = [1, 2, 3]

def generate_grades():
    # 랜덤으로 하나의 학년(grade) 추출
    return random.choice(grades)

• 실질적으로 학생 정보를 20,000건 생성한다
• 생성한 20,000 건의 학생 데이터는 판다스(Pandas)의 데이터프레임(dataframe) 형태로 저장한다

# 학생 정보 생성 함수
def generate_student():
    name = generate_name()
    department = genearte_department()
    score = generate_score()
    grade = generate_grades()

    return name, department, score, grade

# 20,000 건의 학생 데이터 생성 (학번, 이름, 학과, 성적, 학년)
students = []
cnt = 20000
for id in range(1, cnt + 1):
    name, department, score, grade = generate_student()
    students.append((id, name, department, score, grade))

df = pd.DataFrame(students, columns=["id", "name", "department", "score", "grade"])
df.to_csv("students.csv") # 학생 데이터를 엑셀 파일(.csv) 형태로 저장
df.head()

• 이렇게 만든 데이터로 어떤 자료구조가 좋을지 테스트해보자!

1-1. 단순 리스트 사용해보기

• 가장 성적이 낮은 100명의 정보를 추출해보자
• 이어서 가장 성적이 낮은 10,000명의 정보를 추출해보자
• 정렬을 사용하지 않고, 단순히 구현해보자

# 학생 리스트 내에서 가장 성적이 낮은 학생의 인덱스(index) 반환
def get_min(students):
    min_index = 0
    for i in range(len(students)):
        if students[min_index][3] > students[i][3]:
            min_index = i
    return min_index

start_time = time.time()

students = []
df = pd.read_csv("students.csv")

for index, row in df.iterrows():
    id = row["id"]
    name = row["name"]
    department = row["department"]
    score = row["score"]
    grade = row["grade"]
    student = (id, name, department, score, grade)
    students.append(student)

print(f"Data inserted ({time.time() - start_time:.4f} seconds.)")

# 가장 작은 100개 추출
smallest = []
for i in range(100):
    index = get_min(students)
    smallest.append(students[index])
    del students[index]

print("[The five smallest scores]")
for small in smallest[:5]:
    print(small[1])

# 점수가 작은 순서대로 10,000 건의 학생 데이터 추출
for i in range(10000):
    index = get_min(students)
    del students[index]
print(f"Data deleted ({time.time() - start_time:.4f} seconds.)")

# 남은 원소 중에서 가장 점수가 낮은 학생 정보 출력
index = get_min(students)
print(students[index])

1-2. 힙(Heap) 자료구조 사용해보기

• 힙(heap) 자료구조에 학생의 데이터를 성적순으로 삽입하여 관리해보자
• 가장 성적이 낮은 100명의 정보를 추출해보자
• 이어서 가장 성적이 낮은 10,000명의 정보를 추출해보자

start_time = time.time()

heap = []
df = pd.read_csv("students.csv")

for index, row in df.iterrows():
    id = row["id"]
    name = row["name"]
    department = row["department"]
    score = row["score"]
    grade = row["grade"]
    student = (score, (id, name, department, score, grade))
    heapq.heappush(heap, student)

print(f"Data inserted ({time.time() - start_time:.4f} seconds.)")

# 가장 작은 100개 추출
smallest = []
for i in range(100):
    element = heapq.heappop(heap)
    smallest.append(element)

print("[The five smallest scores]")
for small in smallest[:5]:
    print(small[1])

# 점수가 작은 순서대로 10,000 건의 학생 데이터 추출
for i in range(10000):
    heapq.heappop(heap)
print(f"Data deleted ({time.time() - start_time:.4f} seconds.)")

# 남은 원소 중에서 가장 점수가 낮은 학생 정보 출력
print(heap[0])

• 힙이 훨씬 빠르다는 것을 확인할 수 있다!

2. 대량의 데이터를 파일 입출력하기

• 대량의 데이터가 들어있는 파일을 입출력할 때는 파이썬 기본 입력 함수 또는 판다스(Pandas) 라이브러리를 사용할 수 있다
• 어떤게 더 빠를지 테스트해보자

2-1. 대량의 데이터 파일 만들기

• 위에서 사용한 학생 생성을 가지고 1,000명, 10,000명, 100,000명이 들어있는 파일들을 각각 만들어보자

import random
import pandas as pd
import os
import numpy as np
import time
import pandas as pd
from IPython.display import display, Image

last_names = [
    "Smith", "Johnson", "Williams", "Jones", "Brown",
    "Davis", "Miller", "Wilson", "Moore", "Taylor",
    "Anderson", "Thomas", "Jackson", "White", "Harris",
    "Martin", "Thompson", "Garcia", "Martinez", "Robinson",
    "Clark", "Rodriguez", "Lewis", "Lee", "Walker",
    "Hall", "Allen", "Young", "Hernandez", "King"
]

first_names = [
    "James", "Mary", "Robert", "Patricia"	, "John",
    "Jennifer", "Michael", "Linda", "David", "Elizabeth",
    "William", "Barbara", "Richard", "Susan", "Joseph",
    "Jessica", "Thomas", "Sarah", "Charles", "Karen",
    "Christopher", "Lisa", "Daniel", "Nancy", "Matthew",
    "Betty", "Anthony", "Margaret", "Mark", "Sandra"
]

departments = [
    "Computer Science",
    "Mechanical Engineering",
    "Biomedical Engineering",
    "Radiology",
    "Psychology"
]

mu = 50
sigma = 10

grades = [1, 2, 3]

def generate_name():
    # 랜덤으로 하나의 성(last name) 추출
    last_name = random.choice(last_names)
    # 랜덤으로 하나의 이름(first name) 추출
    first_name = random.choice(first_names)
    return first_name + " " + last_name

def genearte_department():
    # 랜덤으로 하나의 학과(department) 추출
    return random.choice(departments)

def generate_score():
    # 랜덤으로 하나의 성적(score) 추출
    return np.random.normal(mu, sigma, 1)[0]

def generate_grades():
    # 랜덤으로 하나의 학년(grade) 추출
    return random.choice(grades)
    
# 학생 정보 생성 함수
def generate_student():
    name = generate_name()
    department = genearte_department()
    score = generate_score()
    grade = generate_grades()

    return name, department, score, grade

def generate_dataset(cnt, path):
    students = []

    # 다수의 학생 데이터 생성 (학번, 이름, 학과, 성적, 학년)
    for id in range(1, cnt + 1):
        name, department, score, grade = generate_student()
        students.append((id, name, department, score, grade))

    df = pd.DataFrame(students, columns=["id", "name", "department", "score", "grade"])
    df.to_csv(path) # 학생 데이터를 엑셀 파일(.csv) 형태로 저장
    df.head()

generate_dataset(1000, "students_1000.csv")
generate_dataset(10000, "students_10000.csv")
generate_dataset(100000, "students_100000.csv")

result_path = ["students_1000.csv", "students_10000.csv", "students_100000.csv"]

for path in result_path:
    print(f"[File: {path}]")
    n = os.path.getsize(path)
    print(f"Total size: {n:.2f} bytes.")
    print(f"Total size: {n / 1024:.2f} KB.")
    print(f"Total size: {n / 1024 / 1024:.2f} MB.")

2-2. Pandas 라이브러리 사용해보기

• 엑셀 파일(.csv)의 경우 판다스(Pandas) 라이브러리를 이용해 데이터프레임(dataframe) 형태로 불러올 수 있다

def csv_reader(path):
    start_time = time.time()

    students = []
    df = pd.read_csv(path)
    for index, row in df.iterrows():
        id = row["id"]
        name = row["name"]
        department = row["department"]
        score = row["score"]
        grade = row["grade"]
        student = (id, name, department, score, grade)
        students.append(student)

    print(f"Data inserted ({time.time() - start_time:.4f} seconds.)")
    return students

students = csv_reader("students_100000.csv")
print(students[0])
print(students[99999])

2-3. 파이썬 기본 파일 입출력 함수 사용해보기

• 파이썬의 기본적인 파일 입력 함수를 이용해 데이터를 불러올 수 있다

def file_reader(path):
    start_time = time.time()

    students = []
    with open(path, "r") as f:
        rows = f.readlines()
        # 첫째 줄은 속성(property)에 대한 내용이므로, 둘째 줄부터 읽기
        for row in rows[1:]:
            data = row.strip().split(",")
            index = data[0]
            id = data[1]
            name = data[2]
            department = data[3]
            score = data[4]
            grade = data[5]
            student = (id, name, department, score, grade)
            students.append(student)

    print(f"Data inserted ({time.time() - start_time:.4f} seconds.)")
    return students

students = file_reader("students_100000.csv")
print(students[0])
print(students[99999])

• 경우에 따라서 기본적인 파일 입력 함수를 이용해 직접 구현한 것의 속도가 더 빠를 수 있다는 것을 알 수 있다

3. 누락된 데이터 처리하기

• 누락된 데이터를 다루는 방법은 다양하며, 대표적인 사례는 다음과 같다
  
  1. 값을 0으로 대입하는 방법
     
     • df.fillna(0) : 누락된 데이터를 0으로 채우기
  2. 해당 특징(feature)의 평균 값을 대입하는 방법
     
     • df.fillna(df.mean()) : 누락된 데이터를 해당 열(column)의 평균으로 채우기

3-1. Pandas로 누락된 데이터 처리하기

• dropna() : NaN 값이 하나라도 포함된 행(row)을 삭제한다
• dropna(how="all") : 모든 열의 값이 NaN인 행(row)을 삭제한다
• dropna(how="any") : dropna()과 동일하다
• dropna(thresh=5) : 누락된 데이터를 제외한 열이 5개 이상이면 해당 행(row)을 남긴다
• fillna() : 누락된 데이터를 원하는 값으로 채운다

import random
import pandas as pd
import os
import numpy as np
import time
import pandas as pd

df = pd.DataFrame([
    [98, np.nan, 88],
    [73, np.nan, np.nan],
    [92, 71, 82],
    [np.nan, np.nan, np.nan],
    [72, 91, 78]
], columns=["Math", "Science", "English"])

print(df)

# NaN 값이 하나라도 포함된 행(row)을 삭제
processed = df.dropna()
print(processed)

# 모든 열의 값이 NaN인 행(row)을 삭제
processed = df.dropna(how="all")
print(processed)

# 누락된 데이터를 제외한 열이 2개 이상이면 해당 행(row)을 남김
processed = df.dropna(thresh=2)
print(processed)

# 누락된 데이터를 0으로 채우기
processed = df.fillna(0)
print(processed)

# 누락된 데이터를 해당 열(column)의 평균으로 채우기
processed = df.fillna(df.mean())
print(processed)

# 누락된 데이터를 처리한 뒤에도 원본 df의 값은 유지
print(df)

4. 클래스 활용하기

• 클래스의 상속를 이용해서 간단한 게임 캐릭터 클래스를 만들어보자

class Character:
    def __init__(self, name, hp, strength, agility):
        self.name = name
        self.hp = hp
        self.strength = strength
        self.agility = agility

    def show_character(self):
        print("===== 캐릭터 정보 =====")
        print(f"이름: {self.name}")
        print(f"체력: {self.hp}")
        print(f"힘: {self.strength}")
        print(f"민첩도: {self.agility}")

    def attack(self):
        print(f"[{self.name}] 기본 공격 수행 (공격력: {self.strength})")


character1 = Character("슬라임", 50, 5, 3)
character1.show_character()
character1.attack()

character2 = Character("용사1", 200, 10, 5)
character2.show_character()
character2.attack()

• 그냥 이렇게만 만들면 재미가 없으니 용사에는 직업을 추가하고, 몬스터에는 마나를 추가해서 스킬을 사용할 수 있게 만들어보자

class Monster(Character):
    def __init__(self, name, hp, strength, agility, mp):
        super().__init__(name, hp, strength, agility)
        self.mp = mp

    def recovery(self):
        print(f"[{self.name}] 자기 치유 (회복된 체력: {self.mp})")
        self.hp += self.mp

monster1 = Monster("슬라임", 50, 5, 3, 5)
monster1.show_character()
monster1.attack()
monster1.recovery()
monster1.show_character()

class Hero(Character):
    def __init__(self, name, hp, strength, agility, job):
        super().__init__(name, hp, strength, agility)
        self.job = job

    def show_hero(self):
        print("===== 주인공 정보 =====")
        print(f"직업: {self.job}")

hero1 = Hero("용사1", 200, 10, 5, "마법사")
hero1.show_character()
hero1.show_hero()
hero1.attack()

4-1. 딥러닝 모델 클래스 예시

• 딥러닝 분야에서는 클래스가 어떻게 쓰일 수 있는지 알아보자

4-1-1. 딥러닝 이미지 모델 클래스 (부모)

• 기초적인 이미지 처리 모델은 일반적으로 인코더 정보를 포함한다
• 인코더(encoder)
  
  1. 특징 추출기(feature extractor)
  2. 입력 모양(input shape)
• 딥러닝 모델은 입력을 통해 출력을 뱉는 forward() 함수가 존재한다

class Model:
    def __init__(self, feature_extractor, input_shape):
        self.feature_extractor = feature_extractor
        self.input_shape = input_shape

    def show(self):
        print("===== 인코더 정보 =====")
        print(f"특징 추출기: {self.feature_extractor}")
        print(f"입력 차원: {self.input_shape}")
    
    def forward(self, x):
        print(f"입력 데이터: {x}")
        print(f"[특징 추출기] {self.feature_extractor}")


model = Model("ResNet50 backbone", (224, 224, 3))
model.show()
model.forward(x=None)

 

4-1-2. 이미지 분류 모델 (자식)

• 기초적인 이미지 분류 모델은 다음의 두 가지를 포함하는 경우가 많다
  
  1. 분류기(classifier)
  2. 출력 모양(output shape)
• 오버라이딩(overriding)을 이용해 forward() 함수를 재정의할 수 있다

class Classifier(Model):
    def __init__(self, feature_extractor, input_shape, classifier, output_shape):
        super().__init__(feature_extractor, input_shape)
        self.classifier = classifier
        self.output_shape = output_shape

    def show_classifier(self):
        print("===== 모델 정보 =====")
        print(f"특징 추출기: {self.feature_extractor}")
        print(f"분류 모델: {self.classifier}")
        print(f"입력 차원: {self.input_shape}")
        print(f"출력 차원: {self.output_shape}")

    def forward(self, x):
        print(f"입력 데이터: {x}")
        print(f"[특징 추출기] {self.feature_extractor}")
        print(f"[분류 모델] {self.classifier}")

model = Classifier("ResNet50 backbone", (256, 256, 3), "FC layer", (10))
model.show_classifier()
model.forward(x=None)

 

4-1-3. 이미지 분할(Segmentation) 모델 (자식)

• 기초적인 이미지 분할 모델은 다음의 두 가지를 포함하는 경우가 많다
  
  1. 디코더(decoder)
  2. 출력 모양(output shape)
• 오버라이딩(overriding)을 이용해 forward() 함수를 재정의할 수 있다

class SegmentationModel(Model):
    def __init__(self, feature_extractor, input_shape, decoder, output_shape):
        super().__init__(feature_extractor, input_shape)
        self.decoder = decoder
        self.output_shape = output_shape

    def show_segmentation_model(self):
        print("===== 모델 정보 =====")
        print(f"특징 추출기: {self.feature_extractor}")
        print(f"디코더: {self.decoder}")
        print(f"입력 차원: {self.input_shape}")
        print(f"출력 차원: {self.output_shape}")

    def forward(self, x):
        print(f"입력 데이터: {x}")
        print(f"[특징 추출기] {self.feature_extractor}")
        print(f"[디코더] {self.decoder}")

model = SegmentationModel("ResNet50 backbone", (256, 256, 3), "U-Net", (256, 256, 3))
model.show_segmentation_model()
model.forward(x=None)

 

 

 

 

 

 

 

---