Use FG in MaxCompute tasks - - Centre de documentation Alibaba Cloud

Méthode 1 : Utiliser une image de groupe de ressources à usage général

Dans la console DataWorks, accédez à Scheduling Configuration > Resource Properties. Sélectionnez un groupe de ressources à usage général et choisissez la dernière image dataworks_pairec_task_pod.

Remarque : La publication de l'image dataworks_pairec_task_pod peut être en décalage par rapport aux mises à jour de pyfg ; l'image risque donc de ne pas contenir le dernier package pyfg. Pour vérifier la version requise, consultez le script généré par Recommended Solution Customization - Feature Configuration. Pour utiliser la dernière version de pyfg, utilisez la Method 3 afin de créer une image personnalisée pour votre groupe de ressources.

Méthode 2 : Installer les dépendances (pour les anciennes versions de `DataWorks`)

Connectez-vous à la console DataWorks, créez un groupe de ressources exclusif pour la planification, puis utilisez l'O&M Assistant pour installer le package pyfg.

Pour installer le package pyfg dans votre groupe de ressources exclusif DataWorks, accédez à DataWorks->Management Center->Resource Group List->O&M Assistant et exécutez la commande suivante :

/home/tops/bin/pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple --upgrade --force-reinstall http://tzrec.oss-cn-beijing.aliyuncs.com/third_party/pyfg105-1.0.5-cp37-cp37m-linux_x86_64.whl

FAQ

Si vous rencontrez l'erreur suivante :

Executing user script with PyODPS 0.10.8 (wrapper version: 0.11.6)
/home/tops/lib/python3.7/site-packages/odps/types.py:1381: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  float_builtins += (np.float,)
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:35: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_float_types = list(map(np.dtype, [np.float, np.float32, np.float64]))
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:40: DeprecationWarning: `np.bool` is a deprecated alias for the builtin `bool`. To silence this warning, use `bool` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.bool_` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_to_df_types[np.dtype(np.bool)] = types.boolean
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:41: DeprecationWarning: `np.str` is a deprecated alias for the builtin `str`. To silence this warning, use `str` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.str_` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_to_df_types[np.dtype(np.str)] = types.string
/home/tops/lib/python3.7/site-packages/odps/ipython/magics.py:35: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  np_float_types = map(np.dtype, [np.float, np.float16, np.float32, np.float64])
package: pyfg105
package path: /home/tops/lib/python3.7/site-packages/pyfg105
custom_lib_path: /home/tops/lib/python3.7/site-packages/pyfg105/lib
Initializing pyfg...
WARNING: Logging before InitGoogleLogging() is written to STDERR
I20260525 20:54:34.598881    21 str_utils.cc:252] GlobalConfig ctor: 0x7fc8a7ffa040
I20260525 20:54:34.598944    21 str_utils.cc:54] AVX supported
I20260525 20:54:34.598951    21 str_utils.cc:56] FMA supported
I20260525 20:54:34.598958    21 str_utils.cc:75] AVX-512F not supported
I20260525 20:54:34.598961    21 str_utils.cc:255] support avx512: false
I20260525 20:54:34.598968    21 str_utils.cc:264] will NOT use avx512
I20260525 20:54:34.598982    21 base_feature.cc:705] reset global fg config
pyfg version: 1.0.5 loaded
2026-05-25 20:54:34,606 WARNING:odps.pyodpswrapper:Code with error
===================
from pyfg105 import run_on_odps
fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    force_delete_output_table=True,
    force_update_resource=True)
fg_task.create_output_table(o)
===================
Traceback (most recent call last):
  File "/home/tops/lib/python3.7/site-packages/pyfg105/run_on_odps.py", line 12, in <module>
    from .create_output_table import TableCreator
  File "/home/tops/lib/python3.7/site-packages/pyfg105/create_output_table.py", line 2, in <module>
    from odps.models import TableSchema, Column
ImportError: cannot import name 'TableSchema' from 'odps.models' (/home/tops/lib/python3.7/site-packages/odps/models/__init__.py)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "<pyodps_user_code>", line 1, in <module>
    from pyfg105 import run_on_odps
  File "/home/tops/lib/python3.7/site-packages/pyfg105/run_on_odps.py", line 15, in <module>
    from create_output_table import TableCreator
  File "/home/tops/lib/python3.7/site-packages/pyfg105/create_output_table.py", line 2, in <module>
    from odps.models import TableSchema, Column
ImportError: cannot import name 'TableSchema' from 'odps.models' (/home/tops/lib/python3.7/site-packages/odps/models/__init__.py)

Cette erreur survient car la version de pyodps sur le groupe de ressources exclusif actuel est obsolète. Vous devez la mettre à niveau vers la dernière version et également mettre à jour le package correspondant pyodpswrapper. Toutefois, cette opération risque de modifier les versions d'autres packages dans l'environnement python, ce qui pourrait entraîner l'échec de vos tâches python. L'approche la plus sûre consiste à utiliser une image personnalisée pour votre groupe de ressources (Method 3).

Méthode 3 : Personnaliser une image de groupe de ressources (pour les nouvelles versions de `DataWorks`)

Pour obtenir des instructions, consultez la rubrique Custom images.

Télécharger les fichiers de ressources

Téléchargez le fichier de configuration FG (au format JSON) dans votre projet MaxCompute.

Certains opérateurs de caractéristiques nécessitent des fichiers de ressources supplémentaires. Téléchargez manuellement ces fichiers dans votre projet MaxCompute.

Opérateur de caractéristique	Description	Paramètre du fichier de ressource
text_normalizer	Normalisation du texte	Fichier de mots vides `stop_char_file`
tokenize_feature	Caractéristique de tokenisation de texte	Fichier de configuration du vocabulaire `vocab_file`
bm25_feature	Caractéristique de pertinence textuelle	Fichier de configuration de la fréquence des termes `term_doc_freq_file`
custom_feature	opérateur personnalisé	Fichier de configuration de l'opérateur `operator_lib_file`

Créer la table de sortie

Dans DataWorks, créez un nœud PyOdps3 et exécutez le script suivant. Ce script lit le fichier fg.json et crée la table de sortie ainsi que les ressources nécessaires aux exécutions ultérieures.

from pyfg105 import run_on_odps

fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    force_delete_output_table=True,
    force_update_resource=True)
fg_task.create_output_table(o)

Avant d'exécuter le script, configurez les paramètres suivants dans Scheduling Configuration : input_table, output_table, fg_json_file et partition_value.

Bien que fg_task.run(o) crée automatiquement la table de sortie si elle n'existe pas, nous vous recommandons d'appeler cette méthode pour créer la table de sortie à l'avance. Cela permet d'éviter les conflits et les échecs de tâches lors du remplissage simultané des données.

Exécuter la tâche FG hors ligne

Dans DataWorks, créez un nœud PyOdps3 et exécutez le script suivant. Ce script exécute la tâche de génération de caractéristiques et crée automatiquement la table de sortie si elle n'existe pas.

from pyfg105 import run_on_odps

fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    batch_size=128,
    force_delete_output_table=False,
    force_update_resource=False)
fg_task.add_sql_setting('odps.stage.mapper.split.size', 256)
fg_task.run(o)

Avant d'exécuter le script, configurez les paramètres suivants dans Scheduling Configuration : input_table, output_table, fg_json_file et partition_value.

Si PyODPS est installé, vous pouvez également installer pyfg et soumettre des tâches localement.

Paramètres

Paramètre	Valeur par défaut	Description
input_table	Aucune	La table d'entrée.
output_table	Aucune	La table de sortie. Créée automatiquement si elle n'existe pas.
fg_json_file	Aucune	Le fichier de configuration FG, au format JSON.
partition_value	Aucune	La partition de la table d'entrée à traiter. Les résultats sont écrits dans la partition correspondante de la table de sortie.
schema	Aucune	Le schéma MaxCompute. Pour plus d'informations, consultez les opérations de schéma.
batch_size	128	Le nombre d'enregistrements à traiter dans chaque lot.
memory	1024	La quantité de mémoire allouée au nœud de tâche, en MiB.
force_delete_output_table	False	S'il est défini sur True, le système supprime la table de sortie avant d'exécuter la tâche.
force_update_resource	False	S'il est défini sur True, le système met à jour les ressources avant d'exécuter la tâche. Pour éviter les conflits de concurrence, évitez de laisser ce paramètre défini sur True.
output_merged_str	False	S'il est défini sur True, les chaînes sont automatiquement fusionnées pour produire une grande caractéristique de chaîne au format RTP.
debug	False	S'il est défini sur True, la tâche s'exécute en mode débogage et affiche le contenu de toutes les ressources mises à jour.
sql_setting	Aucune	Paramètres SQL MaxCompute. Utilisez la méthode `fg_task.add_sql_setting` pour les configurer. Pour plus d'informations, consultez les paramètres Flag. Vous pouvez ajouter plusieurs flags.
fg_setting	Aucune	Paramètres FG. Utilisez la méthode `fg_task.add_fg_setting` pour les configurer. Pour plus d'informations, consultez Configuration globale. Vous pouvez ajouter plusieurs éléments de configuration. Disponible depuis la version 0.4.0.

Vous pouvez remplacer les valeurs par défaut des paramètres en les transmettant au constructeur FgTask, comme illustré dans les exemples.

Fonctionnement

Le package pyfg s'exécute sur une machine passerelle au sein d'un groupe de ressources exclusif pour soumettre des tâches SQL à MaxCompute. Vous pouvez également installer pyfg et PyODPS sur une machine locale pour soumettre des tâches.

Les UDF personnalisées dans une tâche SQL nécessitent des ressources telles que la bibliothèque partagée FG, les fichiers de configuration (par exemple, fg.json, les dictionnaires et les bibliothèques d'opérateurs personnalisés) ainsi que les fichiers de code UDF (.py). Toutes ces ressources doivent être téléchargées dans le cluster MaxCompute et stockées dans son système de fichiers distribué. Lors de l'exécution d'une tâche, chaque worker télécharge les ressources requises depuis le système de fichiers distribué et les charge en mémoire.

Certaines ressources, comme la bibliothèque partagée FG et les fichiers de code UDF, sont partagées entre plusieurs tâches. Lorsque force_update_resource=True, le système supprime les ressources originales avant de télécharger les nouvelles. Ce processus crée une fenêtre temporelle susceptible de perturber d'autres tâches en cours d'exécution.