메모리 제약 없는 텍스트 분류 (Out-of-Core Learning)

소개

이 실습은 out-of-core 학습을 사용하여 scikit-learn 을 이용한 텍스트 분류 방법을 보여줍니다. 주요 메모리에 맞지 않는 데이터로부터 학습하는 것이 목표입니다. 이를 위해, partial_fit 메서드를 지원하는 온라인 분류기를 사용하여 예제의 배치로 학습시킵니다. 특징 공간이 시간 경과에 따라 동일하게 유지되도록, 각 예제를 동일한 특징 공간으로 투영하는 HashingVectorizer를 활용합니다. 이는 특히 새로운 특징 (단어) 이 각 배치에 나타날 수 있는 텍스트 분류의 경우에 유용합니다.

VM 팁

VM 시작이 완료되면 왼쪽 상단 모서리를 클릭하여 Notebook 탭으로 전환하여 Jupyter Notebook을 이용할 수 있습니다.

때때로 Jupyter Notebook 이 로드되는 데 몇 초 정도 기다려야 할 수 있습니다. Jupyter Notebook 의 제한으로 인해 작업 검증은 자동화될 수 없습니다.

학습 중 문제가 발생하면 Labby 에 문의하십시오. 세션 후 피드백을 제공하면 문제를 신속하게 해결해 드리겠습니다.

라이브러리 가져오기 및 파서 정의

import itertools
from pathlib import Path
from hashlib import sha256
import re
import tarfile
import time
import sys

import numpy as np
import matplotlib.pyplot as plt
from matplotlib import rcParams

from html.parser import HTMLParser
from urllib.request import urlretrieve
from sklearn.datasets import get_data_home
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.linear_model import SGDClassifier
from sklearn.linear_model import PassiveAggressiveClassifier
from sklearn.linear_model import Perceptron
from sklearn.naive_bayes import MultinomialNB


class ReutersParser(HTMLParser):
    """SGML 파일을 파싱하여 문서를 하나씩 생성하는 유틸리티 클래스입니다."""

    def __init__(self, encoding="latin-1"):
        HTMLParser.__init__(self)
        self._reset()
        self.encoding = encoding

    def handle_starttag(self, tag, attrs):
        method = "start_" + tag
        getattr(self, method, lambda x: None)(attrs)

    def handle_endtag(self, tag):
        method = "end_" + tag
        getattr(self, method, lambda: None)()

    def _reset(self):
        self.in_title = 0
        self.in_body = 0
        self.in_topics = 0
        self.in_topic_d = 0
        self.title = ""
        self.body = ""
        self.topics = []
        self.topic_d = ""

    def parse(self, fd):
        self.docs = []
        for chunk in fd:
            self.feed(chunk.decode(self.encoding))
            for doc in self.docs:
                yield doc
            self.docs = []
        self.close()

    def handle_data(self, data):
        if self.in_body:
            self.body += data
        elif self.in_title:
            self.title += data
        elif self.in_topic_d:
            self.topic_d += data

    def start_reuters(self, attributes):
        pass

    def end_reuters(self):
        self.body = re.sub(r"\s+", r" ", self.body)
        self.docs.append(
            {"title": self.title, "body": self.body, "topics": self.topics}
        )
        self._reset()

    def start_title(self, attributes):
        self.in_title = 1

    def end_title(self):
        self.in_title = 0

    def start_body(self, attributes):
        self.in_body = 1

    def end_body(self):
        self.in_body = 0

    def start_topics(self, attributes):
        self.in_topics = 1

    def end_topics(self):
        self.in_topics = 0

    def start_d(self, attributes):
        self.in_topic_d = 1

    def end_d(self):
        self.in_topic_d = 0
        self.topics.append(self.topic_d)
        self.topic_d = ""

Reuters 문서 스트림 정의

def stream_reuters_documents(data_path=None):
    """Reuters 데이터셋의 문서들을 반복합니다.

    `data_path` 디렉토리가 존재하지 않으면 Reuters 아카이브가 자동으로 다운로드 및 압축 해제됩니다.

    문서는 'body' (문자열), 'title' (문자열), 'topics' (문자열 리스트) 키를 가진 사전으로 표현됩니다.

    """

    DOWNLOAD_URL = (
        "http://archive.ics.uci.edu/ml/machine-learning-databases/"
        "reuters21578-mld/reuters21578.tar.gz"
    )
    ARCHIVE_SHA256 = "3bae43c9b14e387f76a61b6d82bf98a4fb5d3ef99ef7e7075ff2ccbcf59f9d30"
    ARCHIVE_FILENAME = "reuters21578.tar.gz"

    if data_path is None:
        data_path = Path(get_data_home()) / "reuters"
    else:
        data_path = Path(data_path)
    if not data_path.exists():
        """데이터셋을 다운로드합니다."""
        print("데이터셋을 %s에 다운로드합니다 (한 번만)." % data_path)
        data_path.mkdir(parents=True, exist_ok=True)

        def progress(blocknum, bs, size):
            total_sz_mb = "%.2f MB" % (size / 1e6)
            current_sz_mb = "%.2f MB" % ((blocknum * bs) / 1e6)
            if _not_in_sphinx():
                sys.stdout.write("\rdownloaded %s / %s" % (current_sz_mb, total_sz_mb))

        archive_path = data_path / ARCHIVE_FILENAME

        urlretrieve(DOWNLOAD_URL, filename=archive_path, reporthook=progress)
        if _not_in_sphinx():
            sys.stdout.write("\r")

        ## 아카이브가 손상되지 않았는지 확인합니다:
        assert sha256(archive_path.read_bytes()).hexdigest() == ARCHIVE_SHA256

        print("Reuters 데이터셋을 압축 해제 중...")
        tarfile.open(archive_path, "r:gz").extractall(data_path)
        print("완료.")

    parser = ReutersParser()
    for filename in data_path.glob("*.sgm"):
        for doc in parser.parse(open(filename, "rb")):
            yield doc

벡터라이저 설정 및 테스트 세트 분리

## 벡터라이저를 생성하고, 특징 수를 적절한 최대값으로 제한합니다.
vectorizer = HashingVectorizer(decode_error="ignore", n_features=2**18, alternate_sign=False)

## 파싱된 Reuters SGML 파일을 반복하는 반복자.
data_stream = stream_reuters_documents()

## "acq" 클래스와 다른 모든 클래스 간의 이진 분류를 학습합니다.
## "acq"는 Reuters 파일에서 거의 균등하게 분포되어 있기 때문에 선택되었습니다.
## 다른 데이터셋의 경우, 양성 인스턴스의 현실적인 비율로 테스트 세트를 만드는 데 주의해야 합니다.
all_classes = np.array([0, 1])
positive_class = "acq"

## `partial_fit` 메서드를 지원하는 몇 가지 분류기가 있습니다.
partial_fit_classifiers = {
    "SGD": SGDClassifier(max_iter=5),
    "Perceptron": Perceptron(),
    "NB Multinomial": MultinomialNB(alpha=0.01),
    "Passive-Aggressive": PassiveAggressiveClassifier(),
}

## 테스트 데이터 통계
test_stats = {"n_test": 0, "n_test_pos": 0}

## 먼저 정확도를 추정하기 위해 일부 예제를 분리합니다.
n_test_documents = 1000
X_test_text, y_test = get_minibatch(data_stream, 1000)
X_test = vectorizer.transform(X_test_text)
test_stats["n_test"] += len(y_test)
test_stats["n_test_pos"] += sum(y_test)
print("테스트 세트는 %d개의 문서 (%d개의 양성)" % (len(y_test), sum(y_test)))

예제 미니배치를 가져오는 함수 정의

def get_minibatch(doc_iter, size, pos_class=positive_class):
    """예제의 미니배치를 추출하고, 튜플 X_text, y 를 반환합니다.

    참고: 크기는 할당된 주제가 없는 유효하지 않은 문서를 제외하기 전의 크기입니다.

    """
    data = [
        ("{title}\n\n{body}".format(**doc), pos_class in doc["topics"])
        for doc in itertools.islice(doc_iter, size)
        if doc["topics"]
    ]
    if not len(data):
        return np.asarray([], dtype=int), np.asarray([], dtype=int)
    X_text, y = zip(*data)
    return X_text, np.asarray(y, dtype=int)

미니배치 반복을 위한 생성자 함수 정의

def iter_minibatches(doc_iter, minibatch_size):
    """미니배치 생성자."""
    X_text, y = get_minibatch(doc_iter, minibatch_size)
    while len(X_text):
        yield X_text, y
        X_text, y = get_minibatch(doc_iter, minibatch_size)

예제의 미니배치를 반복하고 분류기를 업데이트합니다

## 분류기에 1000 개의 문서로 구성된 미니배치를 제공합니다. 즉, 한 번에 메모리에 최대 1000 개의 문서가 있습니다.
## 문서 배치가 작을수록 부분적 적합 (partial fit) 메서드의 상대적 오버헤드가 커집니다.
minibatch_size = 1000

## Reuters SGML 파일을 파싱하고 문서를 스트림으로 반복하는 data_stream 을 생성합니다.
minibatch_iterators = iter_minibatches(data_stream, minibatch_size)
total_vect_time = 0.0

## 주요 루프: 예제의 미니배치를 반복합니다.
for i, (X_train_text, y_train) in enumerate(minibatch_iterators):
    tick = time.time()
    X_train = vectorizer.transform(X_train_text)
    total_vect_time += time.time() - tick

    for cls_name, cls in partial_fit_classifiers.items():
        tick = time.time()
        ## 현재 미니배치의 예제로 추정자를 업데이트합니다.
        cls.partial_fit(X_train, y_train, classes=all_classes)

        ## 테스트 정확도 통계를 누적합니다.
        cls_stats[cls_name]["total_fit_time"] += time.time() - tick
        cls_stats[cls_name]["n_train"] += X_train.shape[0]
        cls_stats[cls_name]["n_train_pos"] += sum(y_train)
        tick = time.time()
        cls_stats[cls_name]["accuracy"] = cls.score(X_test, y_test)
        cls_stats[cls_name]["prediction_time"] = time.time() - tick
        acc_history = (cls_stats[cls_name]["accuracy"], cls_stats[cls_name]["n_train"])
        cls_stats[cls_name]["accuracy_history"].append(acc_history)
        run_history = (
            cls_stats[cls_name]["accuracy"],
            total_vect_time + cls_stats[cls_name]["total_fit_time"],
        )
        cls_stats[cls_name]["runtime_history"].append(run_history)

        if i % 3 == 0:
            print(progress(cls_name, cls_stats[cls_name]))
    if i % 3 == 0:
        print("\n")

결과 플롯

## 정확도 변화 플롯
plt.figure()
for _, stats in sorted(cls_stats.items()):
    ## 예제 수에 따른 정확도 변화 플롯
    accuracy, n_examples = zip(*stats["accuracy_history"])
    plot_accuracy(n_examples, accuracy, "학습 예제 수 (#)")
    ax = plt.gca()
    ax.set_ylim((0.8, 1))
plt.legend(cls_names, loc="best")

plt.figure()
for _, stats in sorted(cls_stats.items()):
    ## 실행 시간에 따른 정확도 변화 플롯
    accuracy, runtime = zip(*stats["runtime_history"])
    plot_accuracy(runtime, accuracy, "실행 시간 (초)")
    ax = plt.gca()
    ax.set_ylim((0.8, 1))
plt.legend(cls_names, loc="best")

## 적합 시간 플롯
plt.figure()
fig = plt.gcf()
cls_runtime = [stats["total_fit_time"] for cls_name, stats in sorted(cls_stats.items())]

cls_runtime.append(total_vect_time)
cls_names.append("벡터화")
bar_colors = ["b", "g", "r", "c", "m", "y"]

ax = plt.subplot(111)
rectangles = plt.bar(range(len(cls_names)), cls_runtime, width=0.5, color=bar_colors)

ax.set_xticks(np.linspace(0, len(cls_names) - 1, len(cls_names)))
ax.set_xticklabels(cls_names, fontsize=10)
ymax = max(cls_runtime) * 1.2
ax.set_ylim((0, ymax))
ax.set_ylabel("실행 시간 (초)")
ax.set_title("학습 시간")


def autolabel(rectangles):
    """직사각형 위에 텍스트를 자동으로 추가합니다."""
    for rect in rectangles:
        height = rect.get_height()
        ax.text(
            rect.get_x() + rect.get_width() / 2.0,
            1.05 * height,
            "%.4f" % height,
            ha="center",
            va="bottom",
        )
        plt.setp(plt.xticks()[1], rotation=30)


autolabel(rectangles)
plt.tight_layout()
plt.show()

## 예측 시간 플롯
plt.figure()
cls_runtime = []
cls_names = list(sorted(cls_stats.keys()))
for cls_name, stats in sorted(cls_stats.items()):
    cls_runtime.append(stats["prediction_time"])
cls_runtime.append(parsing_time)
cls_names.append("읽기/파싱\n+ 특징 추출")
cls_runtime.append(vectorizing_time)
cls_names.append("해싱\n+ 벡터화")

ax = plt.subplot(111)
rectangles = plt.bar(range(len(cls_names)), cls_runtime, width=0.5, color=bar_colors)

ax.set_xticks(np.linspace(0, len(cls_names) - 1, len(cls_names)))
ax.set_xticklabels(cls_names, fontsize=8)
plt.setp(plt.xticks()[1], rotation=30)
ymax = max(cls_runtime) * 1.2
ax.set_ylim((0, ymax))
ax.set_ylabel("실행 시간 (초)")
ax.set_title("%d개 인스턴스의 예측 시간" % n_test_documents)
autolabel(rectangles)
plt.tight_layout()
plt.show()

요약

이 실험에서는 오프라인 학습을 사용하여 scikit-learn 을 통해 텍스트 분류를 수행하는 방법을 배웠습니다. partial_fit 메서드를 지원하는 온라인 분류기를 사용하여 예제의 배치로 학습시켰습니다. 또한, 특징 공간이 시간 경과에 따라 동일하게 유지되도록 HashingVectorizer 를 활용했습니다. 그런 다음 테스트 세트를 따로 두고 미니배치의 예제들을 반복하여 분류기를 업데이트했습니다. 마지막으로, 정확도 변화와 학습 시간을 시각화하기 위해 결과를 플롯했습니다.