Classification de texte en utilisant l'apprentissage hors-mémoire

Introduction

Ce laboratoire fournit un exemple sur la manière d'utiliser scikit-learn pour la classification de texte en utilisant l'apprentissage hors mémoire. L'objectif est d'apprendre à partir de données qui ne rentrent pas en mémoire principale. Pour y arriver, nous utilisons un classifieur en ligne qui prend en charge la méthode partial_fit, qui sera alimenté avec des lots d'exemples. Pour s'assurer que l'espace de caractéristiques reste le même au fil du temps, nous utilisons un HashingVectorizer qui projettera chaque exemple dans le même espace de caractéristiques. Cela est particulièrement utile dans le cas de la classification de texte où de nouvelles caractéristiques (mots) peuvent apparaître dans chaque lot.

Conseils sur la machine virtuelle

Une fois le démarrage de la machine virtuelle terminé, cliquez dans le coin supérieur gauche pour basculer vers l'onglet Notebook pour accéder à Jupyter Notebook pour la pratique.

Parfois, vous devrez peut-être attendre quelques secondes pour que Jupyter Notebook ait fini de charger. La validation des opérations ne peut pas être automatisée en raison des limitations de Jupyter Notebook.

Si vous rencontrez des problèmes pendant l'apprentissage, n'hésitez pas à demander à Labby. Donnez votre feedback après la session, et nous résoudrons rapidement le problème pour vous.

Skills Graph

Importation des bibliothèques et définition du parseur

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):
    """Classe utilitaire pour analyser un fichier SGML et générer des documents un par un."""

    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 = ""

Définition du flux de documents Reuters

def stream_reuters_documents(data_path=None):
    """Itérer sur les documents de l'ensemble de données Reuters.

    L'archive Reuters sera automatiquement téléchargée et décompressée si
    le répertoire `data_path` n'existe pas.

    Les documents sont représentés sous forme de dictionnaires avec les clés 'body' (str),
    'title' (str), 'topics' (list(str)).

    """

    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():
        """Télécharger l'ensemble de données."""
        print("téléchargement de l'ensemble de données (une fois pour toutes) dans %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")

        ## Vérifier que l'archive n'a pas été altérée :
        assert sha256(archive_path.read_bytes()).hexdigest() == ARCHIVE_SHA256

        print("décompression de l'ensemble de données Reuters...")
        tarfile.open(archive_path, "r:gz").extractall(data_path)
        print("terminé.")

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

Configurer le vectoriseur et réserver un ensemble de test

## Créer le vectoriseur et limiter le nombre de fonctionnalités à un maximum raisonnable
vectorizer = HashingVectorizer(decode_error="ignore", n_features=2**18, alternate_sign=False)

## Itérateur sur les fichiers SGML Reuters analysés.
data_stream = stream_reuters_documents()

## Nous apprenons une classification binaire entre la classe "acq" et toutes les autres.
## "acq" a été choisi car il est plus ou moins régulièrement réparti dans les fichiers Reuters.
## Pour d'autres ensembles de données, il faudrait prendre soin de créer un ensemble de test avec
## une partie réaliste d'instances positives.
all_classes = np.array([0, 1])
positive_class = "acq"

## Voici quelques classifieurs qui prennent en charge la méthode `partial_fit`
partial_fit_classifiers = {
    "SGD": SGDClassifier(max_iter=5),
    "Perceptron": Perceptron(),
    "NB Multinomial": MultinomialNB(alpha=0.01),
    "Passive-Aggressive": PassiveAggressiveClassifier(),
}

## statistiques sur les données de test
test_stats = {"n_test": 0, "n_test_pos": 0}

## Tout d'abord, nous réservons un certain nombre d'exemples pour estimer la précision
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("L'ensemble de test est composé de %d documents (%d positifs)" % (len(y_test), sum(y_test)))

Définir une fonction pour obtenir un mini-lot d'exemples

def get_minibatch(doc_iter, size, pos_class=positive_class):
    """Extraire un mini-lot d'exemples, retourner un tuple X_text, y.

    Note : size est avant d'exclure les documents invalides sans sujets assignés.

    """
    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)

Définir une fonction génératrice pour itérer sur les mini-lots

def iter_minibatches(doc_iter, minibatch_size):
    """Générateur de mini-lots."""
    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)

Itérer sur les mini-lots d'exemples et mettre à jour les classifieurs

## Nous alimenterons le classifieur avec des mini-lots de 1000 documents ; cela signifie
## que nous n'avons en mémoire au plus que 1000 documents à tout moment. Plus le lot
## de documents est petit, plus le surcoût relatif des méthodes de mise à jour partielle est élevé.
minibatch_size = 1000

## Créer le flux de données qui analyse les fichiers SGML Reuters et itère sur
## les documents sous forme de flux.
minibatch_iterators = iter_minibatches(data_stream, minibatch_size)
total_vect_time = 0.0

## Boucle principale : itérer sur les mini-lots d'exemples
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()
        ## mettre à jour l'estimateur avec les exemples du mini-batch actuel
        cls.partial_fit(X_train, y_train, classes=all_classes)

        ## accumuler les statistiques de précision sur le test
        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")

Tracer les résultats

## Tracer l'évolution de la précision
plt.figure()
for _, stats in sorted(cls_stats.items()):
    ## Tracer l'évolution de la précision en fonction du nombre d'exemples
    accuracy, n_examples = zip(*stats["accuracy_history"])
    plot_accuracy(n_examples, accuracy, "exemples d'entraînement (#)")
    ax = plt.gca()
    ax.set_ylim((0.8, 1))
plt.legend(cls_names, loc="best")

plt.figure()
for _, stats in sorted(cls_stats.items()):
    ## Tracer l'évolution de la précision en fonction du temps d'exécution
    accuracy, runtime = zip(*stats["runtime_history"])
    plot_accuracy(runtime, accuracy, "temps d'exécution (s)")
    ax = plt.gca()
    ax.set_ylim((0.8, 1))
plt.legend(cls_names, loc="best")

## Tracer les temps d'ajustement
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("Vectorisation")
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("temps d'exécution (s)")
ax.set_title("Temps d'entraînement")


def autolabel(rectangles):
    """attacher du texte via autolabel sur les 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()

## Tracer les temps de prédiction
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("Lecture/Analyse\n+Extraction de caractéristiques")
cls_runtime.append(vectorizing_time)
cls_names.append("Hachage\n+Vectorisation")

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("temps d'exécution (s)")
ax.set_title("Temps de prédiction (%d instances)" % n_test_documents)
autolabel(rectangles)
plt.tight_layout()
plt.show()

Sommaire

Dans ce laboratoire, nous avons appris à utiliser scikit-learn pour la classification de texte en utilisant l'apprentissage hors-mémoire. Nous avons utilisé un classifieur en ligne qui prend en charge la méthode partial_fit, qui a été alimenté avec des lots d'exemples. Nous avons également utilisé un HashingVectorizer pour nous assurer que l'espace de caractéristiques reste le même au fil du temps. Nous avons ensuite réservé un ensemble de test et itéré sur des mini-lots d'exemples pour mettre à jour les classifieurs. Enfin, nous avons tracé les résultats pour visualiser l'évolution de la précision et les temps d'entraînement.