MaxCompute:Utiliser PyODPS dans DataWorks

Dans un nœud PyODPS, DataWorks fournit une variable globale, odps ou o, comme point d'entrée MaxCompute. Vous n'avez pas besoin de la définir manuellement.

# Check whether the pyodps_iris table exists.
print(o.exist_table('pyodps_iris'))

Le code renvoie True, indiquant que la table pyodps_iris existe.

• Vous pouvez exécuter des instructions SQL dans un nœud PyODPS en utilisant des méthodes telles que execute_sql() ou run_sql(). Ces méthodes prennent principalement en charge les instructions DDL (Data Definition Language) et DML (Data Manipulation Language).

  Remarque

  Les méthodes execute_sql() et run_sql() ne prennent pas en charge toutes les instructions SQL. Pour les instructions autres que DDL ou DML, vous devez utiliser d'autres méthodes. Par exemple, utilisez la méthode run_security_query pour exécuter des instructions GRANT ou REVOKE, et utilisez la méthode run_xflow ou execute_xflow pour exécuter des commandes API.

• Pour lire les résultats de l'exécution SQL dans un nœud PyODPS, utilisez open_reader().

Pour plus d'informations sur les opérations liées à SQL dans un nœud PyODPS, consultez SQL.

Pour des raisons de compatibilité historique, Instance Tunnel n'est pas activé par défaut dans DataWorks. Par conséquent, instance.open_reader utilise l'interface Result par défaut, qui est limitée à 10 000 enregistrements et présente des problèmes avec les types de données complexes. Si votre projet ne dispose pas d'un mécanisme de protection des données et que vous devez itérer sur toutes les données ou lire des types de données complexes tels que Array, activez Instance Tunnel et désactivez la limitation limit.

• Désactivation de la limitation limit au moment de l'exécution

  Utilisez l'instruction suivante pour activer globalement Instance Tunnel et désactiver la restriction 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:
      # You can read all data through Instance Tunnel.
      # You can use reader.count to get the record count.

• Désactivation de la limitation limit pour le Reader actuel uniquement

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

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

Pour plus d'informations sur Instance Tunnel et les limites de lecture des données, consultez Lire les résultats de l'exécution SQL.

Dans un nœud PyODPS, les paramètres de planification vous permettent d'obtenir des valeurs telles que la date métier pour l'exécution d'une tâche. Bien que la définition et l'utilisation des paramètres de planification soient identiques à celles des nœuds SQL, leur référence dans le code diffère.

• Les nœuds SQL utilisent directement une chaîne telle que ${param_name} dans le code.

• Au lieu d'effectuer un remplacement de chaîne pour ${param_name}, DataWorks ajoute un dictionnaire global nommé args avant l'exécution du nœud. Vous pouvez obtenir la valeur d'un paramètre de planification en accédant à ce dictionnaire, par exemple args[param_name].

Par exemple, si vous définissez le paramètre de planification ds=${yyyymmdd} dans le volet Basic properties > Parameters, vous pouvez récupérer ce paramètre dans votre code comme suit :

• Obtenez la valeur du paramètre ds.

  print('ds=' + args['ds'])
  # Returns the time of ds, for example, ds=20161116

• Obtenez les données d'une table dans une partition nommée ds=${yyyymmdd}.

  o.get_table('table_name').get_partition('ds=' + args['ds'])
  # Obtain data from the ds partition of the table_name table.

Pour définir des paramètres d'exécution pour une tâche, utilisez le paramètre hints, qui accepte un dict.

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

Vous pouvez également définir sql.settings globalement. Une fois configurés, ces paramètres sont automatiquement appliqués à chaque exécution.

from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # Hints are added based on the global configuration.

Les nœuds DataWorks sont livrés avec les packages tiers préinstallés suivants.

Si vous avez besoin d'un package qui n'est pas préinstallé, utilisez la méthode load_resource_package pour charger un package tiers à partir d'une ressource MaxCompute. Commencez par créer une archive de package avec pyodps-pack, puis utilisez load_resource_package pour la charger avant d'importer son contenu. Pour plus d'informations sur pyodps-pack, consultez Créer un package tiers pour PyODPS et Utiliser un package tiers dans PyODPS.

Exemple :

1. Utilisez la commande suivante pour empaqueter ipaddress.

   pyodps-pack -o ipaddress-bundle.tar.gz ipaddress

2. Après avoir téléchargé et validé ipaddress-bundle.tar.gz en tant que ressource, vous pouvez utiliser le package ipaddress dans un nœud PyODPS 3 comme suit :

   load_resource_package("ipaddress-bundle.tar.gz")
   import ipaddress

DataWorks limite la taille totale des packages téléchargés à 100 Mo. Pour exclure les packages préinstallés de votre bundle, utilisez le paramètre --exclude de pyodps-pack. Par exemple, la commande suivante exclut numpy et pandas.

pyodps-pack -o bundle.tar.gz --exclude numpy --exclude pandas <YOUR_PACKAGE>

Procédure

1. Accordez à l'autre utilisateur les autorisations de projet requises. Pour plus d'informations, consultez Annexe : Accorder des autorisations à un autre compte.

2. Dans le nœud PyODPS, utilisez la méthode as_account pour changer de compte et créer un nouvel objet de point d'entrée.

   import os
   # Make sure the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your AccessKey ID,
   # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your AccessKey Secret.
   # We recommend that you do not use the AccessKey ID and AccessKey Secret strings directly.
   new_odps = o.as_account(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
   )

3. Vérifiez si le changement de compte a réussi.

   Interrogez les informations sur l'utilisateur actuel en ajoutant l'instruction suivante. Si la sortie affiche l'UID de l'autre utilisateur, le changement de compte a réussi.

   print(new_odps.get_project().current_user)

   Remarque

   new_odps représente l'objet de point d'entrée pour le nouveau compte.

Exemple

1. Créez une table et importez des données.

   1. Téléchargez le jeu de données Iris iris.data et renommez-le en iris.csv.

   2. Créez une table nommée pyodps_iris et téléchargez le jeu de données iris.csv. Pour les instructions, consultez Créer une table et télécharger des données.

      Voici l'instruction de création de table.

      CREATE TABLE if not exists pyodps_iris
      (
      sepallength  DOUBLE comment 'sepal length (cm)',
      sepalwidth   DOUBLE comment 'sepal width (cm)',
      petallength  DOUBLE comment 'petal length (cm)',
      petalwidth   DOUBLE comment 'petal width (cm)',
      name         STRING comment 'species'
      );

2. Créez un nœud ODPS SQL pour accorder les autorisations nécessaires. Pour plus d'informations, consultez Annexe : Accorder des autorisations à un autre compte.

3. Créez un nœud PyODPS pour changer de compte. Le code suivant fournit un exemple. Pour plus d'informations, consultez Développer une tâche PyODPS 3.

   from odps import ODPS
   import os
   import sys
   # Make sure the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to the user's AccessKey ID,
   # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to the user's AccessKey Secret.
   # We recommend that you do not use the AccessKey ID and AccessKey Secret strings directly.
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'] = '<AccessKey ID>'
   os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'] = '<AccessKey Secret>'
   od = o.as_account(os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'), 
            os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
           )
   iris = DataFrame(od.get_table('pyodps_iris'))
   # Print the output by using a conditional statement.
   with od.execute_sql('select * from pyodps_iris WHERE sepallength > 5 ').open_reader() as reader4:
       print(reader4.raw)
       for record in reader4:
           print(record["sepallength"],record["sepalwidth"],record["petallength"],record["petalwidth"],record["name"])
   # Print the UID of the current user.
   print(od.get_project().current_user)

4. Exécutez le code et affichez le résultat.

   Executing user script with PyODPS 0.11.4.post0
   "sepallength","sepalwidth","petallength","petalwidth","name"
   5.4,3.9,1.7,0.4,"Iris-setosa"
   5.4,3.7,1.5,0.2,"Iris-setosa"
   5.8,4.0,1.2,0.2,"Iris-setosa"
   5.7,4.4,1.5,0.4,"Iris-setosa"
   5.4,3.9,1.3,0.4,"Iris-setosa"
   5.1,3.5,1.4,0.3,"Iris-setosa"
   5.7,3.8,1.7,0.3,"Iris-setosa"
   5.1,3.8,1.5,0.3,"Iris-setosa"
   5.4,3.4,1.7,0.2,"Iris-setosa"
   5.1,3.7,1.5,0.4,"Iris-setosa"
   5.1,3.3,1.7,0.5,"Iris-setosa"
   5.2,3.5,1.5,0.2,"Iris-setosa"
   5.2,3.4,1.4,0.2,"Iris-setosa"
   5.4,3.4,1.5,0.4,"Iris-setosa"
   5.2,4.1,1.5,0.1,"Iris-setosa"
   5.5,4.2,1.4,0.2,"Iris-setosa"
   5.5,3.5,1.3,0.2,"Iris-setosa"
   5.1,3.4,1.5,0.2,"Iris-setosa"
   5.1,3.8,1.9,0.4,"Iris-setosa"
   5.1,3.8,1.6,0.2,"Iris-setosa"
   5.3,3.7,1.5,0.2,"Iris-setosa"
   7.0,3.2,4.7,1.4,"Iris-versicolor"
   6.4,3.2,4.5,1.5,"Iris-versicolor"
   6.9,3.1,4.9,1.5,"Iris-versicolor"
   5.5,2.3,4.0,1.3,"Iris-versicolor"
   6.5,2.8,4.6,1.5,"Iris-versicolor"
   5.7,2.8,4.5,1.3,"Iris-versicolor"
   6.3,3.3,4.7,1.6,"Iris-versicolor"
   6.6,2.9,4.6,1.3,"Iris-versicolor"
   5.2,2.7,3.9,1.4,"Iris-versicolor"
   5.9,3.0,4.2,1.5,"Iris-versicolor"
   6.0,2.2,4.0,1.0,"Iris-versicolor"
   6.1,2.9,4.7,1.4,"Iris-versicolor"
   5.6,2.9,3.6,1.3,"Iris-versicolor"
   6.7,3.1,4.4,1.4,"Iris-versicolor"
   5.6,3.0,4.5,1.5,"Iris-versicolor"
   5.8,2.7,4.1,1.0,"Iris-versicolor"
   6.2,2.2,4.5,1.5,"Iris-versicolor"
   5.6,2.5,3.9,1.1,"Iris-versicolor"
   5.9,3.2,4.8,1.8,"Iris-versicolor"
   6.1,2.8,4.0,1.3,"Iris-versicolor"
   6.3,2.5,4.9,1.5,"Iris-versicolor"
   6.1,2.8,4.7,1.2,"Iris-versicolor"
   6.4,2.9,4.3,1.3,"Iris-versicolor"
   6.6,3.0,4.4,1.4,"Iris-versicolor"
   6.8,2.8,4.8,1.4,"Iris-versicolor"
   6.7,3.0,5.0,1.7,"Iris-versicolor"
   6.0,2.9,4.5,1.5,"Iris-versicolor"
   5.7,2.6,3.5,1.0,"Iris-versicolor"
   5.5,2.4,3.8,1.1,"Iris-versicolor"
   5.5,2.4,3.7,1.0,"Iris-versicolor"
   5.8,2.7,3.9,1.2,"Iris-versicolor"
   6.0,2.7,5.1,1.6,"Iris-versicolor"
   5.4,3.0,4.5,1.5,"Iris-versicolor"
   6.0,3.4,4.5,1.6,"Iris-versicolor"
   6.7,3.1,4.7,1.5,"Iris-versicolor"
   6.3,2.3,4.4,1.3,"Iris-versicolor"
   5.6,3.0,4.1,1.3,"Iris-versicolor"
   5.5,2.5,4.0,1.3,"Iris-versicolor"
   5.5,2.6,4.4,1.2,"Iris-versicolor"
   6.1,3.0,4.6,1.4,"Iris-versicolor"
   5.8,2.6,4.0,1.2,"Iris-versicolor"
   5.6,2.7,4.2,1.3,"Iris-versicolor"
   5.7,3.0,4.2,1.2,"Iris-versicolor"
   5.7,2.9,4.2,1.3,"Iris-versicolor"
   6.2,2.9,4.3,1.3,"Iris-versicolor"
   5.1,2.5,3.0,1.1,"Iris-versicolor"
   5.7,2.8,4.1,1.3,"Iris-versicolor"
   6.3,3.3,6.0,2.5,"Iris-virginica"
   5.8,2.7,5.1,1.9,"Iris-virginica"
   7.1,3.0,5.9,2.1,"Iris-virginica"
   6.3,2.9,5.6,1.8,"Iris-virginica"
   6.5,3.0,5.8,2.2,"Iris-virginica"
   7.6,3.0,6.6,2.1,"Iris-virginica"
   7.3,2.9,6.3,1.8,"Iris-virginica"
   6.7,2.5,5.8,1.8,"Iris-virginica"
   7.2,3.6,6.1,2.5,"Iris-virginica"
   6.5,3.2,5.1,2.0,"Iris-virginica"
   6.4,2.7,5.3,1.9,"Iris-virginica"
   6.8,3.0,5.5,2.1,"Iris-virginica"
   5.7,2.5,5.0,2.0,"Iris-virginica"
   5.8,2.8,5.1,2.4,"Iris-virginica"
   6.4,3.2,5.3,2.3,"Iris-virginica"
   6.5,3.0,5.5,1.8,"Iris-virginica"
   7.7,3.8,6.7,2.2,"Iris-virginica"
   7.7,2.6,6.9,2.3,"Iris-virginica"
   6.0,2.2,5.0,1.5,"Iris-virginica"
   6.9,3.2,5.7,2.3,"Iris-virginica"
   5.6,2.8,4.9,2.0,"Iris-virginica"
   7.7,2.8,6.7,2.0,"Iris-virginica"
   6.3,2.7,4.9,1.8,"Iris-virginica"
   6.7,3.3,5.7,2.1,"Iris-virginica"
   7.2,3.2,6.0,1.8,"Iris-virginica"
   6.2,2.8,4.8,1.8,"Iris-virginica"
   6.1,3.0,4.9,1.8,"Iris-virginica"
   6.4,2.8,5.6,2.1,"Iris-virginica"
   7.2,3.0,5.8,1.6,"Iris-virginica"
   7.4,2.8,6.1,1.9,"Iris-virginica"
   7.9,3.8,6.4,2.0,"Iris-virginica"
   6.4,2.8,5.6,2.2,"Iris-virginica"
   6.3,2.8,5.1,1.5,"Iris-virginica"
   6.1,2.6,5.6,1.4,"Iris-virginica"
   7.7,3.0,6.1,2.3,"Iris-virginica"
   6.3,3.4,5.6,2.4,"Iris-virginica"
   6.4,3.1,5.5,1.8,"Iris-virginica"
   6.0,3.0,4.8,1.8,"Iris-virginica"
   6.9,3.1,5.4,2.1,"Iris-virginica"
   6.7,3.1,5.6,2.4,"Iris-virginica"
   6.9,3.1,5.1,2.3,"Iris-virginica"
   5.8,2.7,5.1,1.9,"Iris-virginica"
   6.8,3.2,5.9,2.3,"Iris-virginica"
   6.7,3.3,5.7,2.5,"Iris-virginica"
   6.7,3.0,5.2,2.3,"Iris-virginica"
   6.3,2.5,5.0,1.9,"Iris-virginica"
   6.5,3.0,5.2,2.0,"Iris-virginica"
   6.2,3.4,5.4,2.3,"Iris-virginica"
   5.9,3.0,5.1,1.8,"Iris-virginica"
   <User 139xxxxxxxxxxxxx>

# -*- dump_traceback: true -*-

-- Add another Alibaba Cloud account user at the project level.
add user ALIYUN$<account_name>;
-- Grant the CreateInstance permission on the project.
grant CreateInstance on project <project_name> to USER ALIYUN$<account_name>;
-- Grant the Describe and Select permissions on the table.
grant Describe, Select on table <table_name> to USER ALIYUN$<account_name>;
-- View the authorization results.
show grants for ALIYUN$<account_name>;

Authorization Type: ACL
[user/ALIYUN$:xxx.com]
A    projects/mc_xxx : CreateInstance
A    projects/mc_xxx/tables/pyodps_xxx : Describe | Select