PyTorch Dataset · DataLoader · MLP 학습: 데이터를 배치로 나누어 학습 파이프라인 만들기
데이터를 한꺼번에 모델에 넣지 않고 작은 배치로 나누어 MLP를 학습한다. PyTorch의 Dataset과 DataLoader가 샘플 조회, 섞기, 배치 묶기를 어떻게 나누어 담당하는지 익힌다. 학습과 평가 모드를 구분하고 배치 크기에 따른 속도와 손실 곡선 차이도 비교한다.
By 송민성
오늘의 목표
작은 분류 데이터를 Dataset으로 감싸고 DataLoader로 배치 단위 공급한다. 각 배치로 다층 퍼셉트론(Multilayer Perceptron, 여러 선형층과 활성화 함수를 연결한 신경망)을 학습하고 평가한다.
개념과 직관
데이터를 작은 묶음으로 나누는 이유
샘플 1,000개가 있다고 하자. 모델에 데이터를 전달하는 방법은 세 가지다.
배치 크기 1
[샘플 1] → 학습
[샘플 2] → 학습
...
배치 크기 16
[샘플 1~16] → 학습
[샘플 17~32] → 학습
...
배치 크기 1,000
[샘플 1~1,000] → 학습배치(batch, 한 번의 계산에 함께 사용하는 샘플 묶음)를 사용하면 여러 샘플의 계산을 동시에 처리할 수 있다.
미니배치(mini-batch, 전체 데이터보다 작은 배치)를 이용한 학습은 다음 과정을 반복한다.
데이터에서 배치 꺼내기
↓
모델 예측
↓
손실 계산
↓
기울기 계산
↓
파라미터 갱신배치 하나를 처리할 때마다 파라미터를 한 번 갱신한다.
한 에포크의 갱신 횟수 ≈ 전체 샘플 수 / 배치 크기에포크(epoch)는 전체 학습 데이터를 한 번 사용한 단위다. 샘플이 128개라면 배치 크기 1은 약 128번, 배치 크기 16은 약 8번, 배치 크기 128은 1번 파라미터를 갱신한다.
Dataset
데이터셋(Dataset, 샘플을 저장하고 번호로 꺼내는 규칙을 정의한 객체)은 데이터 창고의 선반과 같다.
PyTorch Dataset에는 두 메서드가 필요하다.
def __len__(self):
# 전체 샘플 수를 반환
def __getitem__(self, index):
# index번째 샘플을 반환그림으로 보면 다음과 같다.
Dataset
├── 0번 → (입력, 정답)
├── 1번 → (입력, 정답)
├── 2번 → (입력, 정답)
└── ...__len__은 선반에 물건이 몇 개 있는지 알려준다. __getitem__은 지정한 번호의 물건을 꺼낸다.
DataLoader
데이터 로더(DataLoader, Dataset에서 샘플을 꺼내 배치로 공급하는 객체)는 창고 직원을 닮았다.
Dataset: 샘플 보관과 조회
DataLoader: 조회 순서 결정과 배치 구성주요 설정은 다음과 같다.
batch_size: 한 배치에 넣을 샘플 수shuffle=True: 에포크마다 샘플 순서를 섞음shuffle=False: 저장된 순서대로 샘플을 꺼냄
학습 데이터는 보통 shuffle=True로 설정한다. 매번 같은 순서로 학습해서 순서에 치우치는 일을 줄이기 위해서다. 평가 데이터는 파라미터를 바꾸지 않으므로 보통 섞지 않는다.
collate
콜레이트(collate, 여러 샘플을 하나의 배치로 묶는 과정)는 낱개 샘플을 상자 하나에 포장한다.
Dataset이 샘플 세 개를 다음처럼 반환한다고 하자.
([1.0, 2.0], 0)
([3.0, 4.0], 1)
([5.0, 6.0], 1)DataLoader의 기본 collate는 이를 다음 모양으로 묶는다.
입력 배치:
[[1.0, 2.0],
[3.0, 4.0],
[5.0, 6.0]]
정답 배치:
[0, 1, 1]입력 길이가 모두 같다면 기본 collate로 충분하다. 문장처럼 길이가 서로 다른 데이터는 그대로 쌓을 수 없다. 이후 패딩(padding, 짧은 입력 뒤에 빈 값을 채워 길이를 맞추는 것)을 배울 때 사용자 정의 collate_fn을 사용하게 된다.
model.train()과 model.eval()
model.train()은 모델을 학습 모드로 전환한다. model.eval()은 평가 모드로 전환한다.
현재 만들 MLP는 두 모드에서 계산 결과가 같지만, 이후 드롭아웃(Dropout, 학습 중 일부 값을 무작위로 끄는 층) 같은 층을 사용하면 동작이 달라진다.
학습 시작 전: model.train()
평가 시작 전: model.eval()model.eval()은 기울기 계산을 끄지 않는다. 평가할 때는 torch.no_grad()도 함께 사용한다.
model.eval()
with torch.no_grad():
prediction = model(x)역할은 서로 다르다.
model.eval() → 일부 층의 동작 방식을 평가용으로 변경
torch.no_grad() → 기울기 계산과 저장을 중단코드로 직접 해보기
1. 배치가 만들어지는 원리 확인하기
먼저 파이썬 리스트를 직접 잘라 배치를 만든다.
numbers = list(range(10))
batch_size = 4
for start in range(0, len(numbers), batch_size):
batch = numbers[start:start + batch_size]
print(batch)실행 결과:
[0, 1, 2, 3]
[4, 5, 6, 7]
[8, 9]마지막 배치는 데이터가 부족하므로 크기가 2다. DataLoader도 기본적으로 마지막의 작은 배치를 유지한다.
2. 작은 분류 데이터 만들기
점의 두 좌표가 같은 부호면 1, 다른 부호면 0으로 분류한다.
(양수, 양수) → 1
(음수, 음수) → 1
(양수, 음수) → 0
(음수, 양수) → 0직선 하나로 나누기 어려운 형태이므로 은닉층이 있는 MLP를 사용한다.
import torch
torch.manual_seed(42)
x = torch.randn(512, 2)
y = ((x[:, 0] * x[:, 1]) > 0).float().unsqueeze(1)
print(x.shape) # torch.Size([512, 2])
print(y.shape) # torch.Size([512, 1])
print(x[:3])
print(y[:3])unsqueeze(1)은 정답 모양을 [512]에서 [512, 1]로 바꾼다. 모델 출력과 정답 모양을 맞추기 위한 처리다.
3. Dataset 구현하기
from torch.utils.data import Dataset
class PointDataset(Dataset):
def __init__(self, features, labels):
self.features = features
self.labels = labels
def __len__(self):
return len(self.features)
def __getitem__(self, index):
return self.features[index], self.labels[index]직접 호출해 동작을 확인한다.
dataset = PointDataset(x, y)
print(len(dataset))
sample_x, sample_y = dataset[0]
print(sample_x)
print(sample_y)dataset[0]을 실행하면 파이썬이 내부적으로 dataset.__getitem__(0)을 호출한다.
4. DataLoader로 배치 만들기
from torch.utils.data import DataLoader
loader = DataLoader(
dataset,
batch_size=16,
shuffle=True,
)
batch_x, batch_y = next(iter(loader))
print(batch_x.shape) # torch.Size([16, 2])
print(batch_y.shape) # torch.Size([16, 1])DataLoader는 반복 가능한 객체다. for문을 사용하면 모든 배치를 순서대로 꺼낼 수 있다.
for batch_x, batch_y in loader:
print(batch_x.shape, batch_y.shape)마지막 배치는 데이터 수에 따라 다른 크기가 될 수 있다. 코드에서 배치 크기를 항상 16이라고 가정하면 안 된다.
5. MLP 정의하기
from torch import nn
class MLP(nn.Module):
def __init__(self):
super().__init__()
self.layers = nn.Sequential(
nn.Linear(2, 16),
nn.ReLU(),
nn.Linear(16, 1),
)
def forward(self, x):
return self.layers(x)데이터가 두 좌표를 가지므로 첫 층의 입력 크기는 2다. 마지막 층은 두 분류 중 클래스 1에 해당할 점수 하나를 출력한다.
6. 배치 단위로 학습하기
다음 코드는 데이터 생성부터 학습과 평가까지 한 번에 실행된다.
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
torch.manual_seed(42)
class PointDataset(Dataset):
def __init__(self, features, labels):
self.features = features
self.labels = labels
def __len__(self):
return len(self.features)
def __getitem__(self, index):
return self.features[index], self.labels[index]
class MLP(nn.Module):
def __init__(self):
super().__init__()
self.layers = nn.Sequential(
nn.Linear(2, 16),
nn.ReLU(),
nn.Linear(16, 1),
)
def forward(self, x):
return self.layers(x)
x = torch.randn(512, 2)
y = ((x[:, 0] * x[:, 1]) > 0).float().unsqueeze(1)
train_x, test_x = x[:400], x[400:]
train_y, test_y = y[:400], y[400:]
train_dataset = PointDataset(train_x, train_y)
test_dataset = PointDataset(test_x, test_y)
train_loader = DataLoader(
train_dataset,
batch_size=16,
shuffle=True,
)
test_loader = DataLoader(
test_dataset,
batch_size=64,
shuffle=False,
)
model = MLP()
loss_fn = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
for epoch in range(30):
model.train()
total_loss = 0.0
for batch_x, batch_y in train_loader:
logits = model(batch_x)
loss = loss_fn(logits, batch_y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item() * len(batch_x)
average_loss = total_loss / len(train_dataset)
if epoch % 5 == 0:
print(f"epoch={epoch:2d}, loss={average_loss:.4f}")
model.eval()
correct = 0
with torch.no_grad():
for batch_x, batch_y in test_loader:
logits = model(batch_x)
probability = torch.sigmoid(logits)
prediction = (probability >= 0.5).float()
correct += (prediction == batch_y).sum().item()
accuracy = correct / len(test_dataset)
print(f"test accuracy={accuracy:.3f}")BCEWithLogitsLoss는 이진 분류 손실 함수다. 모델이 출력한 점수에 필요한 변환과 손실 계산을 함께 처리한다. 학습 중에는 sigmoid를 별도로 적용하지 않는다.
한 배치의 학습 순서는 다음과 같다.
logits = model(batch_x)
loss = loss_fn(logits, batch_y)
optimizer.zero_grad()
loss.backward()
optimizer.step()평균 손실을 계산할 때는 배치 손실에 실제 배치 크기를 곱한다.
total_loss += loss.item() * len(batch_x)
average_loss = total_loss / len(train_dataset)마지막 배치가 더 작을 수 있기 때문이다.
오늘의 실험
배치 크기 1, 16, 128을 사용해 손실 곡선과 실행 시간을 비교한다. 다른 조건은 같게 유지한다.
import time
import torch
from torch import nn
from torch.utils.data import DataLoader
def run_experiment(batch_size):
torch.manual_seed(42)
loader = DataLoader(
train_dataset,
batch_size=batch_size,
shuffle=True,
)
model = MLP()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
loss_fn = nn.BCEWithLogitsLoss()
history = []
start = time.perf_counter()
for epoch in range(30):
model.train()
total_loss = 0.0
for batch_x, batch_y in loader:
logits = model(batch_x)
loss = loss_fn(logits, batch_y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item() * len(batch_x)
history.append(total_loss / len(train_dataset))
elapsed = time.perf_counter() - start
model.eval()
with torch.no_grad():
logits = model(test_x)
prediction = (torch.sigmoid(logits) >= 0.5).float()
accuracy = (prediction == test_y).float().mean().item()
return history, elapsed, accuracy
for batch_size in [1, 16, 128]:
history, elapsed, accuracy = run_experiment(batch_size)
print(
f"batch={batch_size:3d}, "
f"time={elapsed:.3f}s, "
f"final_loss={history[-1]:.4f}, "
f"accuracy={accuracy:.3f}"
)
print("loss:", [round(history[i], 4) for i in [0, 4, 9, 19, 29]])결과를 다음 표에 기록한다.
| 배치 크기 | 1 에포크 손실 | 5 에포크 손실 | 30 에포크 손실 | 실행 시간 | 평가 정확도 |
|---|---|---|---|---|---|
| 1 | |||||
| 16 | |||||
| 128 |
관찰할 내용:
- 배치 크기 1은 한 에포크마다 파라미터를 자주 갱신한다.
- 배치 크기 128은 한 에포크의 갱신 횟수가 적다.
- 큰 배치가 항상 더 높은 정확도를 만들지는 않는다.
- 작은 데이터와 CPU에서는 시간 차이가 실행 환경에 따라 달라진다.
- 공정한 비교를 위해 난수 시드, 모델 구조, 에포크 수, 학습률을 같게 유지한다.
배치 크기를 바꾸면 한 에포크의 파라미터 갱신 횟수도 바뀐다. 따라서 같은 에포크 수 비교와 같은 갱신 횟수 비교는 서로 다른 실험이다.
선택 실험으로 손실 곡선을 그릴 수 있다.
import matplotlib.pyplot as plt
for batch_size in [1, 16, 128]:
history, _, _ = run_experiment(batch_size)
plt.plot(history, label=f"batch={batch_size}")
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.show()왜 이걸 배우나 (LLM / Post-training 연결)
LLM 사후학습(Post-training, 사전학습된 모델을 특정 목적에 맞게 추가 학습하는 과정)에서도 Dataset이 학습 샘플을 정의하고 DataLoader가 토큰화된 샘플을 배치로 묶는다. LLM Research Engineer는 배치 크기, 데이터 섞기, collate 규칙, 학습·평가 모드가 실험 결과와 메모리 사용량에 어떤 영향을 주는지 확인해야 한다.
초보자가 막히는 지점
__len__과 __getitem__ 이름을 틀림
밑줄이 앞뒤로 두 개씩 필요하다.
def __len__(self):
return len(self.features)
def __getitem__(self, index):
return self.features[index], self.labels[index]입력과 정답의 샘플 수가 다름
Dataset을 만들기 전에 첫 번째 차원을 확인한다.
print(features.shape)
print(labels.shape)
assert len(features) == len(labels)출력과 정답의 모양이 다름
모델 출력이 [16, 1]인데 정답이 [16]이면 의도하지 않은 계산이나 경고가 생길 수 있다.
print(logits.shape)
print(batch_y.shape)정답을 만들 때 열 차원을 추가한다.
y = y.unsqueeze(1)학습 데이터에 shuffle=False를 사용함
코드는 실행되지만 매 에포크 같은 순서로 학습한다.
train_loader = DataLoader(
train_dataset,
batch_size=16,
shuffle=True,
)평가 중에도 파라미터를 갱신함
평가에서는 backward()와 optimizer.step()을 호출하지 않는다.
model.eval()
with torch.no_grad():
logits = model(test_x)model.eval()이 기울기를 끈다고 오해함
model.eval()과 torch.no_grad()는 대체 관계가 아니다. 평가할 때 둘을 함께 사용한다.
BCEWithLogitsLoss 앞에 sigmoid를 적용함
BCEWithLogitsLoss가 필요한 변환을 내부에서 처리한다.
# 학습
logits = model(batch_x)
loss = loss_fn(logits, batch_y)
# 평가 확률
probability = torch.sigmoid(logits)마지막 배치 크기를 고정값으로 가정함
샘플 수가 배치 크기로 나누어떨어지지 않으면 마지막 배치는 작다. 현재 배치 크기는 len(batch_x)로 확인한다.
알고리즘 (병행 1시간)
쉬움: LeetCode 455번 Assign Cookies
탐욕법(greedy algorithm, 현재 단계에서 가장 유리해 보이는 선택을 반복하는 방법)으로 해결한다.
아이의 만족 기준과 쿠키 크기를 각각 정렬한다. 가장 작은 쿠키로 만족시킬 수 있는 가장 요구가 작은 아이부터 배정한다.
아이 요구: [1, 2, 3]
쿠키 크기: [1, 1, 3]
쿠키 1 → 요구 1인 아이에게 배정
쿠키 1 → 요구 2를 만족시키지 못함
쿠키 3 → 요구 2인 아이에게 배정
결과: 2명시간복잡도는 정렬 때문에 O(n log n + m log m)이다.
중간: LeetCode 55번 Jump Game
각 위치를 지나며 현재까지 도달할 수 있는 가장 먼 인덱스를 기록한다.
가장 먼 위치 = max(기존 가장 먼 위치, 현재 위치 + 현재 점프 길이)현재 위치가 가장 먼 도달 위치보다 크다면 그 위치에는 도착할 수 없으므로 실패다. 마지막 인덱스까지 도달 범위가 확장되면 성공이다.
시간복잡도는 배열을 한 번 순회하므로 O(n), 추가 공간복잡도는 O(1)이다.
각 문제마다 다음 세 항목을 기록한다.
- 접근: 매 단계에서 어떤 선택을 했고 그 선택이 이전 선택을 망치지 않는 이유
- 시간복잡도: 정렬과 순회가 각각 몇 번 필요한지
- 틀린 이유: 정렬 누락, 포인터 이동 오류, 도달 불가능한 위치 검사 누락 중 무엇이었는지