DataWorks:Développer une tâche PyODPS 3

Prérequis

Un nœud PyODPS 3 a été créé. Pour plus d'informations, consultez la rubrique Créer et gérer des nœuds MaxCompute.

Informations contextuelles

PyODPS est le SDK Python pour MaxCompute. Il offre une interface de programmation Python permettant de rédiger des tâches MaxCompute, d'interroger des tables et des vues, ainsi que de gérer les ressources. Pour en savoir plus, reportez-vous à la documentation PyODPS. Dans DataWorks, vous pouvez utiliser un nœud PyODPS pour planifier et exécuter des tâches Python, puis les intégrer à d'autres types de tâches.

Remarques relatives à l'utilisation

• Si votre code PyODPS nécessite des packages tiers, installez-les en utilisant un groupe de ressources serverless et une image personnalisée.

  Remarque

  Cette méthode n'est pas prise en charge si votre code inclut une fonction définie par l'utilisateur (UDF) qui fait référence à un package tiers. Pour connaître la configuration appropriée, consultez la rubrique Exemple UDF : Utiliser des packages tiers dans les UDF Python.

• Pour mettre à niveau la version de PyODPS, utilisez une image personnalisée afin d'exécuter la commande /home/tops/bin/pip3 install pyodps==0.12.1 sur un groupe de ressources serverless (remplacez 0.12.1 par la version cible de PyODPS). Vous pouvez également utiliser l'Assistant O&M pour exécuter la même commande sur un groupe de ressources exclusif dédié à la planification.

• Si votre tâche PyODPS doit accéder à un environnement réseau spécifique, tel qu'une source de données ou un service situé dans un VPC ou dans un centre de données sur site (IDC), utilisez un groupe de ressources serverless et établissez une connexion réseau entre ce groupe et l'environnement cible. Pour plus de détails, consultez les Solutions de connectivité réseau.

• Pour obtenir des informations sur la syntaxe PyODPS, consultez la documentation PyODPS.

• Les nœuds PyODPS existent en deux variantes : PyODPS 2 et PyODPS 3. Elles s'appuient sur des versions différentes de Python : les nœuds PyODPS 2 utilisent Python 2, tandis que les nœuds PyODPS 3 utilisent Python 3. Veillez à créer le type de nœud correspondant à votre version de Python.

• Si l'exécution du SQL dans un nœud PyODPS ne parvient pas à générer correctement la linéarité des données, empêchant ainsi son affichage dans Data Map, résolvez le problème en définissant manuellement les paramètres de planification et d'exécution de DataWorks dans votre code. Pour apprendre à visualiser la linéarité des données, consultez la rubrique Visualiser la linéarité des données. Pour la configuration des paramètres, reportez-vous à la section Définir des indications de paramètre d'exécution. Vous pouvez obtenir les paramètres d'exécution requis à l'aide de l'exemple de code suivant :

  import os
  ...
  # get DataWorks sheduler runtime parameters
  skynet_hints = {}
  for k, v in os.environ.items():
      if k.startswith('SKYNET_'):
          skynet_hints[k] = v
  ...
  # setting hints while submiting a task
  o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)
  ...

• La taille maximale du journal de sortie d'un nœud PyODPS est limitée à 4 Mo. Évitez d'imprimer de grands volumes de résultats de données dans le journal. Contentez-vous d'afficher les informations essentielles relatives aux alertes et à la progression.

Limites

• Lors de l'exécution d'un nœud PyODPS sur un groupe de ressources exclusif dédié à la planification, ne traitez pas plus de 50 Mo de données locales. Cette restriction découle des spécifications matérielles du groupe de ressources exclusif. Le traitement d'un volume important de données locales dépassant le seuil du système d'exploitation peut provoquer une erreur de mémoire insuffisante (OOM), signalée par le message Got Killed. Évitez d'écrire directement dans le nœud PyODPS un code effectuant un traitement intensif des données.

• Lorsque vous exécutez un nœud PYODPS à l'aide d'un groupe de ressources serverless, vous pouvez configurer les unités de calcul (CU) du nœud en fonction du volume de données à traiter.

  Remarque

  Lors de l'exécution d'une tâche dans un groupe de ressources serverless, une seule tâche prend en charge une configuration maximale de 64CU. Toutefois, il est recommandé de ne pas dépasser 16CU afin d'éviter les pénuries de ressources dues à une valeur de CU excessive, lesquelles pourraient impacter le démarrage de la tâche.

• Une erreur Got killed indique que l'utilisation de la memory a dépassé la limite, entraînant la terminaison du processus. Pour prévenir ce problème, évitez les opérations sur les données locales. Cette limitation ne s'applique pas aux tâches SQL ou DataFrame (à l'exception de to_pandas) initiées via PyODPS.

• Vous pouvez utiliser les bibliothèques Numpy et Pandas préinstallées pour le code ne faisant pas appel à des fonctions personnalisées. Les autres packages tiers contenant du code binaire ne sont pas pris en charge.

• Pour des raisons de compatibilité, l'option options.tunnel.use_instance_tunnel est définie sur False par défaut dans DataWorks. Si vous devez activer globalement le tunnel d'instance, vous devez définir manuellement cette valeur sur True.

• La définition du bytecode diffère selon les versions mineures de Python 3, telles que Python 3.8 et Python 3.7.

  MaxCompute utilise actuellement Python 3.7. Une erreur d'exécution se produira si vous employez une syntaxe propre à d'autres versions de Python 3, comme le bloc finally de Python 3.8. Nous vous recommandons d'utiliser Python 3.7.

• PyODPS 3 prend en charge l'exécution sur un groupe de ressources serverless. Pour en acheter et en utiliser un, consultez la rubrique Utiliser des groupes de ressources serverless.

• L'exécution simultanée de plusieurs tâches Python au sein d'un seul nœud PyODPS n'est pas prise en charge.

Modifier le code : Exemple de base

Après avoir créé un nœud PyODPS, vous pouvez modifier et exécuter votre code. Pour plus d'informations sur la syntaxe PyODPS, consultez la rubrique Aperçu des opérations de base.

• Point d'entrée ODPS

  Un nœud PyODPS DataWorks fournit une variable globale, nommée odps ou o, servant de point d'entrée ODPS. Vous n'avez pas besoin de la définir manuellement.

  print(odps.exist_table('PyODPS_iris'))

• Exécuter du SQL

  Vous pouvez exécuter des instructions SQL dans un nœud PyODPS. Pour plus d'informations, consultez la rubrique SQL.

  • Par défaut, le tunnel d'instance est désactivé dans DataWorks. Cela signifie que instance.open_reader utilise l'interface Result, qui lit au maximum 10 000 enregistrements. Vous pouvez utiliser reader.count pour obtenir le nombre d'enregistrements. Pour parcourir toutes les données, vous devez désactiver la limit. Utilisez les instructions suivantes pour activer globalement le tunnel d'instance et désactiver la limit.

    options.tunnel.use_instance_tunnel = True
    options.tunnel.limit_instance_tunnel = False  # Disable the limit to read all data.
    with instance.open_reader() as reader:
      # All data can be read through the instance tunnel.

  • Vous pouvez également ajouter tunnel=True à open_reader afin d'activer le tunnel d'instance pour l'appel actuel de open_reader. De même, ajoutez limit=False pour désactiver la restriction de limit lors de l'appel en cours.

    # Use the Instance Tunnel interface for the current open_reader operation to read all data.
    with instance.open_reader(tunnel=True, limit=False) as reader:

• Paramètres d'exécution

  • Définissez les paramètres d'exécution à l'aide du paramètre hints, qui est un dict. Pour plus d'informations sur les hints, consultez la rubrique Opérations SET.

    o.execute_sql('select * from PyODPS_iris', hints={'odps.sql.mapper.split.size': 16})

  • Si vous définissez sql.settings dans la configuration globale, ces paramètres d'exécution seront ajoutés à chaque exécution.

    from odps import options
    options.sql.settings = {'odps.sql.mapper.split.size': 16}
    o.execute_sql('select * from PyODPS_iris')  # This call includes hints from the global configuration.

• Résultats d'exécution

  Une instance d'exécution SQL peut directement effectuer l'opération open_reader dans les deux scénarios suivants :

  • L'instruction SQL renvoie des données structurées.

    with o.execute_sql('select * from dual').open_reader() as reader:
    	for record in reader:  # Process each record.

  • Lors de l'exécution d'instructions telles que desc, vous pouvez récupérer le résultat brut de l'exécution SQL en utilisant la propriété reader.raw.

    with o.execute_sql('desc dual').open_reader() as reader:
    	print(reader.raw)

    Remarque

    Si vous utilisez des paramètres de planification personnalisés, vous devez coder en dur l'heure lorsque vous déclenchez directement l'exécution d'un nœud PyODPS 3 depuis la page. Le nœud PyODPS ne peut pas substituer directement cette valeur.

• DataFrame

  Vous pouvez également traiter les données à l'aide d'un DataFrame (non recommandé).

  • Exécution

    Dans l'environnement DataWorks, les opérations DataFrame doivent être explicitement déclenchées en appelant une méthode exécutée immédiatement.

    from odps.df import DataFrame
    iris = DataFrame(o.get_table('pyodps_iris'))
    for record in iris[iris.sepal_width < 3].execute():  # Call an immediately executed method to process each record.

    Si vous devez déclencher une exécution immédiate lors de l'affichage, vous devez activer options.interactive.

    from odps import options
    from odps.df import DataFrame
    options.interactive = True  # Enable the switch at the beginning.
    iris = DataFrame(o.get_table('pyodps_iris'))
    print(iris.sepal_width.sum())  # An immediate execution is triggered when printing.

  • Afficher des informations détaillées

    Définissez l'option options.verbose. Cette option est activée par défaut dans DataWorks et affiche des informations détaillées, telles que l'URL Logview, pendant l'exécution.

Exemple

L'exemple suivant illustre l'utilisation d'un nœud PyODPS :

1. Préparez l'ensemble de données et créez la table d'exemple pyodps_iris. Pour plus de détails, consultez la rubrique Traiter les données à l'aide de DataFrame.

2. Créez un DataFrame. Pour plus de détails, consultez la rubrique Créer un DataFrame à partir d'une table MaxCompute.

3. Saisissez et exécutez le code suivant dans le nœud PyODPS.

   from odps.df import DataFrame
   # Create a DataFrame from an ODPS table.
   iris = DataFrame(o.get_table('pyodps_iris'))
   print(iris.sepallength.head(5))

   Le résultat suivant est renvoyé :

      sepallength
   0          4.5
   1          5.5
   2          4.9
   3          5.0
   4          6.0

Modifier le code : Exemple avancé

Si le nœud doit s'exécuter périodiquement, vous devez définir ses propriétés de planification. Pour plus d'informations, consultez la rubrique Configurer les propriétés de planification d'un nœud.

print('ds=' + args['ds'])
ds=20161116

o.get_table('table_name').get_partition('ds=' + args['ds'])

Pour plus d'informations sur le développement de tâches PyODPS pour d'autres cas d'utilisation, consultez les rubriques suivantes :

• Utiliser des packages tiers dans PyODPS.

• Segmenter du texte chinois avec Jieba dans un nœud PyODPS.

• Utiliser un nœud PyODPS pour éviter le téléchargement de données sur un ordinateur local.

Étapes suivantes

• Déterminer si un script Shell personnalisé a bien été exécuté : La logique permettant de déterminer si un script Python personnalisé a bien été exécuté est identique à celle d'un script Shell. Vous pouvez utiliser cette méthode pour la vérification.

• Déployer une tâche : Si vous utilisez un espace de travail en mode standard, vous devez déployer la tâche dans l'environnement de production avant qu'elle ne puisse s'exécuter périodiquement.

• O&M pour les tâches exécutées périodiquement : Une fois la tâche déployée dans l'environnement de production et planifiée, vous pouvez effectuer des opérations d'O&M sur celle-ci dans le Operation Center.

• FAQ PyODPS : Trouvez des réponses aux questions courantes concernant l'exécution des tâches PyODPS afin de vous aider à résoudre rapidement les problèmes.

FAQ

Q : J'utilise un nœud PyODPS3 pour collecter des données depuis une API tierce, telle que Lark, et les importer dans DataWorks. Le code s'exécute sans problème dans mon environnement de développement local, mais signale une erreur de délai d'attente de réponse lorsqu'il est soumis à l'environnement de production et exécuté dans l'Operation Center. Pourquoi ?

R : Dans Management Center > Security Settings > Workspace. Dans la liste blanche du sandbox, ajoutez le nom de domaine de l'API tierce pour accorder l'accès à la tâche PyODPS 3. Par exemple :

Ajoutez le nom de domaine de l'API Lark open.feishu.cn et définissez le port sur 443.