Inférence de modèle à l'aide de Hugging Face Transformers pour le PNL
- Cette documentation a été retirée et pourrait ne pas être mise à jour. Les produits, services ou technologies mentionnés dans ce contenu ne sont plus pris en charge.
- Databricks recommande plutôt d’utiliser
ai_querypour l’inférence par lots. Voir Enrichir des données à l’aide des AI Functions.
Cet article vous montre comment utiliser les Hugging Face Transformers pour l'inférence de modèle en traitement du langage naturel (TLN).
Hugging Face Transformers fournit la classe pipelines pour utiliser le modèle pré-entraîné pour l'inférence. 🤗 Les pipelines de Transformers prennent en charge une large gamme de tâches de PNL que vous pouvez facilement utiliser sur Databricks.
Exigences
- MLflow 2.3
- N'importe quel cluster sur lequel la bibliothèque
transformersHugging Face est installée peut être utilisé pour l'inférence par batch. La bibliothèquetransformersest préinstallée sur Databricks Runtime 10.4 LTS ML et versions ultérieures. De nombreux modèles NLP populaires fonctionnent mieux sur du matériel GPU ; vous obtiendrez donc les meilleures performances en utilisant du matériel GPU récent, à moins que vous n'utilisiez un modèle spécifiquement optimisé pour une utilisation sur des CPU.
Utilisez les UDF Pandas pour distribuer le calcul de modèle sur un cluster Spark
Lorsque vous faites des expérimentations avec des modèles pré-entraînés, vous pouvez utiliser les UDF Pandas pour encapsuler le modèle et effectuer des calculs sur des CPU ou GPU worker. Les UDF Pandas distribuent le modèle à chaque worker.
Vous pouvez également créer un pipeline Hugging Face Transformers pour la traduction automatique et utiliser une UDF Pandas pour exécuter le pipeline sur les Worker d'un cluster Spark :
import pandas as pd
from transformers import pipeline
import torch
from pyspark.sql.functions import pandas_udf
device = 0 if torch.cuda.is_available() else -1
translation_pipeline = pipeline(task="translation_en_to_fr", model="t5-base", device=device)
@pandas_udf('string')
def translation_udf(texts: pd.Series) -> pd.Series:
translations = [result['translation_text'] for result in translation_pipeline(texts.to_list(), batch_size=1)]
return pd.Series(translations)
La définition du device de cette manière garantit que les GPU sont utilisés s'ils sont disponibles sur le cluster.
Les pipelines Hugging Face pour la traduction renvoient une liste d'objets Python dict, chacun avec une seule clé translation_text et une valeur contenant le texte traduit. Cette UDF extrait la traduction des résultats pour renvoyer une série Pandas avec uniquement le texte traduit. Si votre pipeline a été configuré pour utiliser des GPUs en définissant device=0, alors Spark réaffecte automatiquement les GPUs sur les nœuds worker si votre cluster dispose d'instances avec plusieurs GPUs.
Pour utiliser l'UDF pour traduire une colonne de texte, vous pouvez appeler l'UDF dans une instruction select :
texts = ["Hugging Face is a French company based in New York City.", "Databricks is based in San Francisco."]
df = spark.createDataFrame(pd.DataFrame(texts, columns=["texts"]))
display(df.select(df.texts, translation_udf(df.texts).alias('translation')))
Retourner des types de résultats complexes
En utilisant les UDF Pandas, vous pouvez également renvoyer une sortie plus structurée. Par exemple, dans la reconnaissance d’entités nommées, les pipelines renvoient une liste d’objets dict contenant l’entité, son étendue, son type et un score associé. Bien que similaire à l'exemple de traduction, le type de retour pour l'annotation @pandas_udf est plus complexe dans le cas de la reconnaissance d'entités nommées.
Vous pouvez avoir une idée des types de retour à utiliser en examinant les résultats du pipeline, par exemple en exécutant le pipeline sur le driver.
Dans cet exemple, utilisez le code suivant :
from transformers import pipeline
import torch
device = 0 if torch.cuda.is_available() else -1
ner_pipeline = pipeline(task="ner", model="Davlan/bert-base-multilingual-cased-ner-hrl", aggregation_strategy="simple", device=device)
ner_pipeline(texts)
Pour obtenir les annotations :
[[{'entity_group': 'ORG',
'score': 0.99933606,
'word': 'Hugging Face',
'start': 0,
'end': 12},
{'entity_group': 'LOC',
'score': 0.99967843,
'word': 'New York City',
'start': 42,
'end': 55}],
[{'entity_group': 'ORG',
'score': 0.9996372,
'word': 'Databricks',
'start': 0,
'end': 10},
{'entity_group': 'LOC',
'score': 0.999588,
'word': 'San Francisco',
'start': 23,
'end': 36}]]
Pour représenter cela comme un type de retour, vous pouvez utiliser un array de struct champs, en listant les dict entrées comme les champs du struct:
import pandas as pd
from pyspark.sql.functions import pandas_udf
@pandas_udf('array<struct<word string, entity_group string, score float, start integer, end integer>>')
def ner_udf(texts: pd.Series) -> pd.Series:
return pd.Series(ner_pipeline(texts.to_list(), batch_size=1))
display(df.select(df.texts, ner_udf(df.texts).alias('entities')))
Optimiser les performances
Plusieurs aspects clés sont à prendre en compte pour l'ajustement des performances de l'UDF. Le premier consiste à utiliser chaque GPU efficacement, ce que vous pouvez ajuster en modifiant la taille des batchs envoyés au GPU par le pipeline Transformers. Le second consiste à s'assurer que le DataFrame est bien partitionné pour utiliser l'intégralité du cluster.
Enfin, vous pouvez souhaiter mettre en cache le modèle Hugging Face pour économiser le temps de chargement du modèle ou les coûts d'ingestion.
Choisissez une taille de batch
Bien que les UDFs décrits ci-dessus devraient fonctionner directement avec un batch_size de 1, cela pourrait ne pas utiliser efficacement les Ressources disponibles pour les Worker. Pour améliorer les performances, ajustez la taille du batch au modèle et au matériel dans le cluster. Databricks recommande d'essayer différentes tailles de batch pour le pipeline sur votre cluster afin de trouver les meilleures performances. En savoir plus sur le traitement par lots des pipelines et d'autres options de performance dans la documentation Hugging Face.
Essayez de trouver une taille de batch suffisamment grande pour optimiser l'utilisation complète du GPU, mais qui n'entraîne pas d'erreurs CUDA out of memory. Lorsque vous recevez CUDA out of memory erreurs pendant le réglage, vous devez start une nouvelle session pour libérer la mémoire utilisée par le modèle et les données dans le GPU.
Surveillez les performances du GPU en affichant les mesures de cluster en direct pour un cluster, et en choisissant une métrique, telle que gpu0-util pour l'utilisation du processeur GPU ou gpu0_mem_util pour l'utilisation de la mémoire GPU.
Affiner le parallélisme avec la planification au niveau de l'étape
Par default, Spark planifie une tâche par GPU sur chaque machine. Pour augmenter le parallélisme, vous pouvez utiliser la planification au niveau de l'étape pour indiquer à Spark combien de tâches exécuter par GPU. Par exemple, si vous souhaitez que Spark exécute deux tâches par GPU, vous pouvez le spécifier de la manière suivante :
from pyspark.resource import TaskResourceRequests, ResourceProfileBuilder
task_requests = TaskResourceRequests().resource("gpu", 0.5)
builder = ResourceProfileBuilder()
resource_profile = builder.require(task_requests).build
rdd = df.withColumn('predictions', loaded_model(struct(*map(col, df.columns)))).rdd.withResources(resource_profile)
Repartitioner les données pour utiliser tout le matériel disponible
La deuxième considération pour les performances est d'utiliser pleinement le matériel de votre cluster. Généralement, un petit multiple du nombre de GPU sur vos Worker (pour les clusters GPU) ou du nombre de cœurs sur l'ensemble des Worker de votre cluster (pour les clusters CPU) fonctionne bien. Votre DataFrame d'entrée peut déjà disposer de suffisamment de partitions pour tirer parti du parallélisme du cluster. Pour voir le nombre de partitions que le DataFrame contient, utilisez df.rdd.getNumPartitions(). Vous pouvez repartitionner un DataFrame à l'aide de repartitioned_df = df.repartition(desired_partition_count).
Mettre en cache le modèle dans DBFS ou sur les points de montage
Si vous chargez fréquemment un modèle à partir de clusters différents ou redémarrés, vous pouvez également souhaiter mettre en cache le modèle Hugging Face dans le volume racine DBFS ou sur un point de montage. Cela peut réduire les coûts d'ingestion et le temps de chargement du modèle sur un cluster nouveau ou redémarré. Pour ce faire, définissez la variable d'environnement TRANSFORMERS_CACHE dans votre code avant de charger le pipeline.
Par exemple :
import os
os.environ['TRANSFORMERS_CACHE'] = '/dbfs/hugging_face_transformers_cache/'
Alternativement, vous pouvez obtenir des résultats similaires en enregistrant le modèle dans MLflow avec la variante MLflow transformers.
Notebook : Inférence des transformateurs Hugging Face et journalisation MLflow
Pour get start rapidement avec un exemple de code, ce Notebook est un exemple de bout en bout pour la synthèse de texte en utilisant l'inférence des pipelines Hugging Face Transformers et la journalisation MLflow.
Notebook d'inférence de pipelines Hugging Face Transformers
Ressources supplémentaires
Vous pouvez affiner votre modèle Hugging Face avec les guides suivants :
- Préparer les données pour l'affinement des modèles Hugging Face
- Affiner les modèles Hugging Face pour un seul GPU
En savoir plus sur Que sont les Transformers Hugging Face ?