Script PyAlink - Platform For AI - Centre de documentation Alibaba Cloud

Le composant Script PyAlink vous permet d'appeler n'importe quel algorithme Alink en écrivant du code. Utilisez ce composant pour des tâches de classification, de régression et de recommandation. Le script PyAlink s'intègre parfaitement aux autres composants de Machine Learning Designer, ce qui vous permet de créer et de valider des pipelines métier de bout en bout. Cette rubrique décrit l'utilisation du composant Script PyAlink.

Contexte

Vous pouvez utiliser le composant Script PyAlink de deux manières : séparément ou en combinaison avec d'autres composants Machine Learning Designer. Il offre un accès à des centaines de composants Alink et prend en charge la lecture et l'écriture de types de données via le code. Vous pouvez également déployer un PipelineModel généré par le composant en tant que service EAS. Pour plus d'informations, consultez l'exemple suivant : Exemple : Déployer un modèle généré par un script PyAlink en tant que service EAS.

Concepts clés

Avant d'utiliser le composant Script PyAlink, familiarisez-vous avec les concepts clés suivants.

Concept	Description
operator	Dans Alink, un opérateur représente une fonction algorithmique. Les opérateurs se classent en opérateurs par lot (batch) ou en flux (stream). Par exemple, la régression logistique comprend les opérateurs suivants : LogisticRegressionTrainBatchOp : entraîne la régression logistique. LogisticRegressionPredictBatchOp : effectue des prédictions par lot pour la régression logistique. LogisticRegressionPredictStreamOp : effectue des prédictions en flux pour la régression logistique. Connectez les opérateurs à l'aide des méthodes `link` ou `linkFrom`. L'exemple de code suivant illustre cette procédure. `# Define data. data = CsvSourceBatchOp() # Logistic regression training. lrTrain = LogisticRegressionTrainBatchOp() # Logistic regression prediction. LrPredict = LogisticRegressionPredictBatchOp() # Train. data.link(lrTrain) # Predict. LrPredict.linkFrom(lrTrain, data)` Chaque opérateur possède des paramètres. Par exemple, la régression logistique inclut les paramètres suivants. labelCol : nom de la colonne d'étiquette dans la table d'entrée. Paramètre obligatoire de type String. featureCols : tableau contenant les noms des colonnes de caractéristiques. Ce paramètre est de type String[]. La valeur par défaut est NULL, ce qui indique que toutes les colonnes sont sélectionnées. Pour configurer un paramètre, utilisez le préfixe `set` suivi du nom du paramètre en casse mixte (CamelCase). L'exemple de code suivant illustre cette procédure. `lr = LogisticRegressionTrainBatchOp()\ .setFeatureCols(colnames)\ .setLabelCol("label")` Les sources et les puits de données sont des types d'opérateurs particuliers. Une fois définis, connectez-les aux composants algorithmiques à l'aide des méthodes `link` ou `linkFrom`. Alink inclut des sources de données courantes en flux et par lot. L'exemple de code suivant illustre cette procédure. `df_data = pd.DataFrame([ [2, 1, 1], [3, 2, 1], [4, 3, 2], [2, 4, 1], [2, 2, 1], [4, 3, 2], [1, 2, 1], [5, 3, 2] ]) input = BatchOperator.fromDataframe(df_data, schemaStr='f0 int, f1 int, label int') # load data dataTest = input colnames = ["f0","f1"] lr = LogisticRegressionTrainBatchOp().setFeatureCols(colnames).setLabelCol("label") model = input.link(lr) predictor = LogisticRegressionPredictBatchOp().setPredictionCol("pred") predictor.linkFrom(model, dataTest).print()`
pipeline	Vous pouvez également utiliser les algorithmes Alink dans un pipeline, en combinant le traitement des données, la génération de caractéristiques et l'entraînement du modèle dans un seul pipeline pour l'entraînement, la prédiction et les services en ligne. L'exemple de code suivant illustre cette procédure. quantileDiscretizer = QuantileDiscretizer()\ .setNumBuckets(2)\ .setSelectedCols("sepal_length") binarizer = Binarizer()\ .setSelectedCol("petal_width")\ .setOutputCol("bina")\ .setReservedCols("sepal_length", "petal_width", "petal_length", "category")\ .setThreshold(1.); lda = Lda()\ .setPredictionCol("lda_pred")\ .setPredictionDetailCol("lda_pred_detail")\ .setSelectedCol("category")\ .setTopicNum(2)\ .setRandomSeed(0) pipeline = Pipeline()\ .add(binarizer)\ .add(binarizer)\ .add(lda) pipeline.fit(data1) pipeline.transform(data2)
vector	Type de données personnalisé dans Alink qui prend en charge deux formats : vecteur creux (SparseVector) Exemple : `$4$1:0,1 2:0,2`. Le nombre situé entre les signes dollar (`$`) correspond à la longueur du vecteur. Les valeurs situées après le deuxième signe dollar correspondent à des paires index-valeur. vecteur dense (DenseVector) Exemple : `0,1 0,2 0,3`. Cela représente une séquence de valeurs séparées par des espaces. Remarque Dans Alink, si une colonne est de type vecteur, le nom du paramètre est généralement `vectorColName`.

Composants Alink pris en charge par le script PyAlink

Vous pouvez utiliser des centaines de composants Alink dans un script PyAlink, y compris des composants pour le traitement des données, l'ingénierie des caractéristiques et l'entraînement du modèle.

Remarque

Le composant Script PyAlink prend actuellement en charge les pipelines et les composants par lot, mais ne prend pas en charge les composants en flux.

Méthode 1 : Utiliser le script PyAlink séparément

Cette rubrique utilise un exemple de notation de l'ensemble de données movielens avec le modèle ItemCf pour décrire comment utiliser la plateforme Machine Learning Designer et les ressources Alibaba Cloud afin d'exécuter un workflow implémenté avec un script PyAlink. Procédez comme suit.

Accédez à la page Machine Learning Designer et créez un pipeline vierge. Pour plus d'informations, consultez la section Procédure.
Dans la liste des pipelines, sélectionnez le pipeline vierge que vous avez créé et cliquez sur Open.
Dans la zone de recherche de la liste des composants située à gauche, recherchez PyAlink Script et faites glisser PyAlink Script vers le canevas à droite. Un nœud de pipeline nommé PyAlink Script-1 est automatiquement généré sur le canevas.

Sur le canevas, sélectionnez le nœud PyAlink Script-1. Dans le volet de droite, configurez les paramètres sous les onglets Parameter Settings et Execution tuning.

Sous l'onglet Parameter Settings, saisissez votre code. L'exemple de code suivant illustre cette procédure.

from pyalink.alink import *
def main(sources, sinks, parameter):
    PATH = "http://alink-test.oss-cn-beijing.aliyuncs.com/yuhe/movielens/"
    RATING_FILE = "ratings.csv"
    PREDICT_FILE = "predict.csv"
    RATING_SCHEMA_STRING = "user_id long, item_id long, rating int, ts long"
    ratingsData = CsvSourceBatchOp() \
            .setFilePath(PATH + RATING_FILE) \
            .setFieldDelimiter("\t") \
            .setSchemaStr(RATING_SCHEMA_STRING)
    predictData = CsvSourceBatchOp() \
            .setFilePath(PATH + PREDICT_FILE) \
            .setFieldDelimiter("\t") \
            .setSchemaStr(RATING_SCHEMA_STRING)
    itemCFModel = ItemCfTrainBatchOp() \
            .setUserCol("user_id").setItemCol("item_id") \
            .setRateCol("rating").linkFrom(ratingsData);
    itemCF = ItemCfRateRecommender() \
            .setModelData(itemCFModel) \
            .setItemCol("item_id") \
            .setUserCol("user_id") \
            .setReservedCols(["user_id", "item_id"]) \
            .setRecommCol("prediction_score")
    result = itemCF.transform(predictData)
    result.link(sinks[0])
    BatchOperator.execute()

Un script PyAlink prend en charge quatre ports de sortie. Dans le script, utilisez result.link(sinks[0]) pour écrire les données dans le premier port de sortie. Les composants en aval peuvent lire les données du script en se connectant à ce premier port de sortie. Pour plus d'informations, consultez la section Lecture et écriture de différents types de données dans un script PyAlink.

Sous l'onglet Execution tuning, définissez le mode d'exécution et les spécifications du nœud.

Paramètre	Description
Select job running mode	Les modes suivants sont pris en charge : DLC (Single-machine, multi-concurrency) : recommandé pour les tâches comportant de petits ensembles de données et à des fins de débogage et de validation. MaxCompute (Distributed) : recommandé pour les tâches comportant de grands ensembles de données ou pour les tâches de production. Fully-managed Flink : exécute la tâche sur le cluster Fully-managed Flink lié à l'espace de travail.
Number of workers	Ce paramètre est requis uniquement lorsque le job running mode est défini sur MaxCompute (Distributed) ou Fully-managed Flink (Distributed). Ce paramètre spécifie le nombre de nœuds d'exécution. Si vous laissez ce paramètre vide, le système alloue automatiquement les nœuds en fonction des données de la tâche. Par défaut, ce paramètre est vide.
Memory per worker (MB)	Configurez ce paramètre uniquement lorsque le running mode of the job est défini sur MaxCompute (Distributed) ou Fully-managed Flink (Distributed). Le paramètre spécifie la taille de la mémoire d'un seul nœud en Mo. La valeur doit être un entier positif et la valeur par défaut est 8192.
CPU cores per worker	Ce paramètre est requis uniquement lorsque le Job running mode est défini sur MaxCompute (distributed) ou Fully-managed Flink (distributed). Ce paramètre spécifie le nombre de cœurs CPU pour un seul nœud. La valeur doit être un entier positif et est vide par défaut.
Select node specification to run script	Type de ressource du nœud DLC. La valeur par défaut est 2 vCPU + 8 Go de mémoire - ecs.g6.large.

Au-dessus du canevas, cliquez sur Save, puis cliquez sur l'icône Exécuter pour exécuter le script PyAlink.

Une fois la tâche terminée, cliquez avec le bouton droit sur le nœud PyAlink Script-1 sur le canevas et sélectionnez View Data > Output 0 pour afficher les résultats.

Nom de la colonne

Description

user_id

ID de l'utilisateur.

item_id

ID du film.

prediction_score

Indique la préférence de l'utilisateur pour le film. Ce score sert de référence pour les recommandations.

Méthode 2 : Combiner le script PyAlink avec d'autres composants

Les ports d'entrée et de sortie d'un composant Script PyAlink sont identiques à ceux des autres composants algorithmiques de Machine Learning Designer. Connectez-les pour créer un pipeline combiné, comme illustré dans la figure suivante. 组合使用

Lire et écrire des données dans un script PyAlink

Lire des données.
- Lire à partir d'une table MaxCompute : le script lit les données transmises par un composant en amont via un port d'entrée. L'exemple de code suivant illustre cette procédure.
```
train_data = sources[0]
test_data = sources[1]
```
  Dans le code, sources[0] représente la table MaxCompute connectée au premier port d'entrée, et sources[1] représente la table connectée au deuxième port d'entrée. Le composant prend en charge jusqu'à quatre ports d'entrée.
- Lire à partir d'un système de fichiers réseau : le script lit les données à l'aide des composants source d'Alink, tels que CsvSourceBatchOp et AkSourceBatchOp, directement dans le code. Vous pouvez lire les types de fichiers suivants :
  - Lire un fichier partagé sur un réseau via HTTP. L'exemple de code suivant illustre cette procédure :
```
ratingsData = CsvSourceBatchOp() \
            .setFilePath(PATH + RATING_FILE) \
            .setFieldDelimiter("\t") \
            .setSchemaStr(RATING_SCHEMA_STRING)
```
  - Lire un fichier OSS. Commencez par localiser le chemin de stockage temporaire des données pour le pipeline sous l'onglet propriétés du pipeline. Ce chemin, formaté comme suit : oss://<bucket-name>/<path>, est utilisé pour les opérations de lecture/écriture OSS. L'exemple de code suivant illustre cette procédure.
```
model_data = AkSourceBatchOp().setFilePath("oss://xxxxxxxx/model_20220323.ak")
```
Écrire des données.
- Écrire dans une table MaxCompute : le script écrit les données vers un composant en aval via un port de sortie. L'exemple de code suivant illustre cette procédure.
```
result0.link(sinks[0])
result1.link(sinks[1])
BatchOperator.execute()
```
  La ligne result0.link(sinks[0]) écrit les données et les rend accessibles via le premier port de sortie. Vous pouvez écrire dans un maximum de quatre tables de résultats, correspondant à quatre ports de sortie.
- Écrire dans un fichier OSS. Utilisez le chemin OSS défini dans le champ chemin de stockage temporaire des données du pipeline, disponible sous l'onglet propriétés du pipeline. L'exemple de code suivant illustre cette procédure.
```
result.link(AkSinkBatchOp() \
            .setFilePath("oss://xxxxxxxx/model_20220323.ak") \
            .setOverwriteSink(True))
BatchOperator.execute()
```

Exemple : Déployer un modèle en tant que service EAS

Générez le modèle à déployer.

Vous pouvez déployer un modèle en tant que service EAS uniquement s'il s'agit d'un PipelineModel généré par le composant Script PyAlink. Utilisez le code suivant pour générer un fichier PipelineModel. Pour savoir comment exécuter le script, consultez la section Méthode 1 : Utiliser le composant Script PyAlink séparément.

from pyalink.alink import *
def main(sources, sinks, parameter):
    PATH = "http://alink-test.oss-cn-beijing.aliyuncs.com/yuhe/movielens/"
    RATING_FILE = "ratings.csv"
    PREDICT_FILE = "predict.csv"
    RATING_SCHEMA_STRING = "user_id long, item_id long, rating int, ts long"
    ratingsData = CsvSourceBatchOp() \
            .setFilePath(PATH + RATING_FILE) \
            .setFieldDelimiter("\t") \
            .setSchemaStr(RATING_SCHEMA_STRING)
    predictData = CsvSourceBatchOp() \
            .setFilePath(PATH + PREDICT_FILE) \
            .setFieldDelimiter("\t") \
            .setSchemaStr(RATING_SCHEMA_STRING)
    itemCFModel = ItemCfTrainBatchOp() \
            .setUserCol("user_id").setItemCol("item_id") \
            .setRateCol("rating").linkFrom(ratingsData);
    itemCF = ItemCfRateRecommender() \
            .setModelData(itemCFModel) \
            .setItemCol("item_id") \
            .setUserCol("user_id") \
            .setReservedCols(["user_id", "item_id"]) \
            .setRecommCol("prediction_score")
    model = PipelineModel(itemCF)
    model.save().link(AkSinkBatchOp() \
            .setFilePath("oss://<your_bucket_name>/model.ak") \
            .setOverwriteSink(True))
    BatchOperator.execute()

Remplacez <your_bucket_name> par le nom de votre compartiment OSS.

Important

Assurez-vous de disposer des autorisations de lecture pour le chemin de l'ensemble de données configuré dans PATH. Sinon, le composant ne pourra pas s'exécuter.

Générez un fichier de configuration EAS.

Exécutez le script suivant pour écrire la sortie dans un fichier config.json.

# EAS configuration file
import json
# Generate EAS model configuration
model_config = {}
# Schema of the data that EAS receives
model_config['inputDataSchema'] = "id long, movieid long" 
model_config['modelVersion'] = "v0.2"
eas_config = {
    "name": "recomm_demo",
    "model_path": "http://xxxxxxxx/model.ak",
    "processor": "alink_outer_processor",
    "metadata": {
        "instance": 1,
        "memory": 2048,
        "region":"China (Beijing)"
    },
    "model_config": model_config
}
print(json.dumps(eas_config, indent=4))

Paramètres clés du fichier config.json :

name : nom du service de modèle déployé.
model_path : chemin OSS où le fichier PipelineModel est stocké. Remplacez cette valeur par le chemin OSS réel de votre fichier de modèle.

Pour une explication des autres paramètres du fichier config.json, consultez la section Référence des commandes.

Déployez le modèle en tant que service EAS.
Déployez le modèle à l'aide du client eascmd. Pour obtenir des instructions sur la configuration du client, consultez la section Télécharger et configurer le client. Par exemple, sur un système Windows 64 bits, exécutez cette commande :
```
eascmdwin64.exe create config.json
```