DataWorks:Nœud PyODPS 2

Présentation

PyODPS est le SDK Python pour MaxCompute. Il offre une interface de programmation concise qui vous permet d'utiliser Python pour écrire des jobs, interroger des tables et des vues, et gérer les ressources MaxCompute. Pour plus d'informations, consultez la rubrique Présentation de PyODPS. Dans DataWorks, vous pouvez utiliser un nœud PyODPS pour planifier et exécuter des tâches Python, et les intégrer à d'autres jobs.

Remarques

• Si votre code nécessite des packages tiers lors de l'exécution d'un nœud PyODPS sur un groupe de ressources DataWorks, utilisez un groupe de ressources serverless et une image personnalisée pour installer les packages.

  Remarque

  Cette méthode n'est pas prise en charge si votre code utilise une UDF qui fait référence à un package tiers. Pour savoir comment gérer ce scénario, consultez la rubrique Exemple d'UDF : Utiliser des packages tiers dans les UDF Python.

• Si votre tâche PyODPS doit accéder à un environnement réseau spécifique, tel qu'une source de données ou un service dans un VPC ou un centre de données sur site, utilisez un groupe de ressources serverless et reportez-vous aux Solutions de connectivité réseau pour établir la connexion réseau requise.

• Pour plus d'informations sur la syntaxe PyODPS, consultez la documentation PyODPS.

• Les nœuds PyODPS sont disponibles en deux types : PyODPS 2 et PyODPS 3. Ils diffèrent par leur version Python sous-jacente. Les nœuds PyODPS 2 utilisent Python 2, tandis que les nœuds PyODPS 3 utilisent Python 3. Créez le type de nœud correspondant à la version Python que vous utilisez.

• Si l'exécution de SQL dans un nœud PyODPS ne parvient pas à générer la lignée des données, ce qui empêche son affichage dans Data Map, vous pouvez résoudre ce problème en définissant manuellement les paramètres de planification DataWorks dans le code de la tâche. Pour afficher la lignée des données, consultez la rubrique Afficher la lignée des données. Pour plus d'informations sur la configuration des paramètres, consultez la rubrique Définir les indicateurs de paramètres 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)
  # ...

• Le journal de sortie d'un nœud PyODPS prend en charge une taille maximale de 4 Mo. Évitez d'imprimer de grands volumes de données directement dans le journal. Contentez-vous de consigner les avertissements essentiels et les mises à jour de progression.

Limites

• Lorsque vous exécutez un nœud PyODPS sur un groupe de ressources exclusif pour la planification, nous recommandons que les données traitées localement au sein du nœud ne dépassent pas 50 Mo. Cette opération est limitée par les spécifications du groupe de ressources exclusif pour la planification. Le traitement d'une quantité excessive de données locales dépassant le seuil du système d'exploitation peut provoquer une erreur OOM (processus tué). Évitez d'écrire trop de code de traitement de données dans les nœuds PyODPS.

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

  Remarque

  Lorsque vous exécutez cette tâche sur un groupe de ressources serverless, la configuration maximale pour une seule tâche est de 64CU, mais nous vous recommandons de ne pas dépasser 16CU afin d'éviter un manque de ressources dû à une allocation excessive d'unités de calcul, ce qui pourrait affecter le démarrage de la tâche.

• Si vous rencontrez une erreur Got Killed, cela indique que l'utilisation de la mémoire a dépassé la limite et que le processus a été arrêté. Par conséquent, évitez autant que possible les opérations de données locales. Les tâches SQL et DataFrame initiées via PyODPS (à l'exception de to_pandas) ne sont pas soumises à cette limitation.

• Le code non-UDF peut utiliser les packages NumPy et Pandas préinstallés. Les autres packages tiers contenant du code binaire ne sont pas pris en charge.

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

• La version Python sous-jacente pour les nœuds PyODPS 2 est la 2,7.

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

• Pour imprimer des journaux dans un nœud PyODPS, utilisez print. L'utilisation de logger.info n'est pas prise en charge.

Prérequis

Associer un moteur de calcul MaxCompute à votre espace de travail DataWorks.

Procédure

1. Sur la page d'édition du nœud PyODPS 2, effectuez les opérations de développement suivantes.

   Exemples de code PyODPS 2

   Après avoir créé un nœud PyODPS, vous pouvez modifier et exécuter du code. Pour plus d'informations sur la syntaxe PyODPS, consultez la documentation PyODPS. Cette rubrique décrit les cinq exemples de code suivants. Sélectionnez les exemples en fonction de vos besoins métier.

   Point d'entrée ODPS

   Dans les nœuds PyODPS de DataWorks, une variable globale odps ou o est incluse en tant que point d'entrée ODPS. Vous n'avez pas besoin de définir manuellement le point d'entrée ODPS.

   print(odps.exist_table('PyODPS_iris'))

   Exécuter SQL

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

   • Par défaut, instance tunnel n'est pas activé sur DataWorks, ce qui signifie que instance.open_reader utilise l'interface Result par défaut (jusqu'à 10 000 enregistrements). Vous pouvez utiliser reader.count pour obtenir le nombre d'enregistrements. Si vous devez itérer sur toutes les données, vous devez désactiver la restriction limit. Vous pouvez utiliser les instructions suivantes pour activer Instance Tunnel globalement et désactiver la restriction limit.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable limit restriction, read all data.
     
     with instance.open_reader() as reader:
       # Use Instance Tunnel to read all data.

   • Vous pouvez également ajouter tunnel=True à open_reader pour activer instance tunnel uniquement pour l'appel open_reader actuel. Vous pouvez aussi ajouter limit=False pour désactiver la restriction limit uniquement pour l'appel actuel.

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

   Définir les paramètres d'exécution

   • Vous pouvez définir les paramètres d'exécution en configurant le paramètre hints, qui est de type dict. Pour plus d'informations sur le paramètre hints, consultez la rubrique Opérations SET.

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

   • Après avoir configuré sql.settings globalement, vous devez ajouter les paramètres d'exécution pertinents chaque fois que vous exécutez une tâche.

     from odps import options
     options.sql.settings = {'odps.sql.mapper.split.size': 16}
     o.execute_sql('select * from PyODPS_iris')  # Add hints based on global configurationhints。

   Lire les résultats d'exécution

   Une instance qui exécute du SQL peut effectuer directement l'opération open_reader. Deux scénarios sont possibles :

   • 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 eachrecord。

   • L'instruction SQL peut être une instruction telle que desc. Vous pouvez utiliser la propriété reader.raw pour obtenir le résultat brut de l'exécution SQL.

     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 les valeurs temporelles lorsque vous déclenchez directement un nœud PyODPS 2 sur la page, car un nœud PyODPS ne peut pas remplacer directement les paramètres.

   DataFrame

   Vous pouvez également utiliser DataFrame pour traiter les données.

   • Exécution

     Dans l'environnement DataWorks, l'exécution de DataFrame nécessite des appels explicits aux méthodes qui déclenchent l'exécution immédiate.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Execute immediately and process each Record.

     Si vous souhaitez déclencher l'exécution immédiate lors de l'utilisation de Print, 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())  # Executes immediately on print.

   • Imprimer des informations détaillées

     Vous pouvez configurer l'option options.verbose. Sur DataWorks, cette option est activée par défaut, et des informations détaillées telles que l'URL Logview sont imprimées pendant l'exécution.

   Développement de code PyODPS 2

   L'exemple simple suivant montre comment utiliser un nœud PyODPS :

   1. Préparez un jeu de données et créez la table d'exemple pyodps_iris. Pour plus d'informations, consultez la rubrique Créer un DataFrame.

   2. Créez un DataFrame. Pour plus d'informations, consultez la rubrique Créer un DataFrame.

   3. Saisissez le code suivant dans le nœud PyODPS.

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

   Exécuter une tâche PyODPS

   1. Dans Run Configuration, configurez Compute Resource, Compute Resource and Quota et DataWorks Resource Group.

      Remarque

      • Pour accéder aux sources de données dans un environnement de réseau public ou VPC, utilisez un groupe de ressources de planification qui a réussi le test de connectivité pour la source de données. Pour plus d'informations, consultez les Solutions de connectivité réseau.

      • Vous pouvez configurer les informations Image en fonction des exigences de la tâche.

   2. Dans la barre d'outils, cliquez sur Run pour exécuter la tâche PyODPS.

2. Si vous devez exécuter la tâche de nœud de manière périodique, configurez les paramètres de planification en fonction de vos besoins métier. Pour plus d'informations, consultez la rubrique Configurer les paramètres de planification.

   Contrairement aux nœuds SQL dans DataWorks, les nœuds PyODPS ne remplacent pas les chaînes telles que ${param_name} dans le code afin d'éviter d'affecter le code. Au lieu de cela, avant d'exécuter le code, un dict nommé args est ajouté aux variables globales, à partir duquel les paramètres de planification peuvent être récupérés. Par exemple, si vous définissez ds=${yyyymmdd} dans Parameter, vous pouvez récupérer le paramètre dans votre code comme suit.

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

   Remarque

   Si vous devez obtenir la partition nommée ds, vous pouvez utiliser la méthode suivante.

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

3. Après avoir terminé la configuration de la tâche de nœud, vous devez déployer le nœud. Pour plus d'informations, consultez la rubrique Déployer un nœud.

4. Après avoir déployé la tâche, vous pouvez afficher l'état d'exécution de la tâche planifiée dans Operation Center. Pour plus d'informations, consultez la rubrique Afficher les tâches planifiées.

Exécuter un nœud à l'aide d'un rôle associé

Vous pouvez associer un rôle RAM pour exécuter un nœud, ce qui vous permet d'exécuter des tâches de nœud avec un rôle RAM spécifique pour un contrôle d'accès granulaire et une gestion de la sécurité.

Étape suivante

FAQ PyODPS : Découvrez les problèmes courants lors de l'exécution de PyODPS pour un dépannage rapide.