Exécuter et obtenir des résultats - MaxCompute - Centre de documentation Alibaba Cloud

Cette rubrique décrit les méthodes d'exécution disponibles pour les opérations DataFrame.

Prérequis

Vérifiez que les conditions suivantes sont remplies :

Une table d'exemple nommée pyodps_iris est prête. Pour plus d'informations, consultez la section Traitement des données DataFrame.
Un objet DataFrame est créé. Pour plus d'informations, consultez la section « Créer un objet DataFrame à partir d'une table MaxCompute » dans Créer un objet DataFrame.

Exécution différée

Les opérations DataFrame ne s'exécutent que lorsque vous appelez explicitement la méthode execute ou lorsque vous invoquez des méthodes qui appellent implicitement execute. Le tableau suivant répertorie les méthodes qui déclenchent l'exécution.

Méthode	Description	Valeur de retour
persist	Enregistre les résultats d'exécution dans des tables MaxCompute.	PyODPS DataFrame
execute	Exécute les opérations et renvoie tous les résultats.	ResultFrame
head	Exécute les opérations et renvoie les N premières lignes des données de résultat.	ResultFrame
tail	Exécute les opérations et renvoie les N dernières lignes des données de résultat.	ResultFrame
to_pandas	Convertit un objet Collection en objet pandas DataFrame ou un objet Sequence en objet Series. Si le paramètre wrap est défini sur True, un objet PyODPS DataFrame est renvoyé.	Si le paramètre wrap est défini sur True, un objet PyODPS DataFrame est renvoyé. Si le paramètre wrap est défini sur False, un objet pandas DataFrame est renvoyé. La valeur par défaut du paramètre wrap est False.
plot, hist et boxplot	Méthodes de tracé.	N/A

Remarque

Dans un environnement interactif, PyODPS DataFrame appelle automatiquement la méthode execute lors de l'affichage des données de résultat ou de l'appel à la méthode repr. Vous n'avez pas besoin d'appeler manuellement la méthode execute.

Exemples

# In a non-interactive environment, you need to manually call the execute method.
print(iris[iris.sepallength < 5][:5].execute())

# In an interactive environment, the system automatically calls the execute method.
print(iris[iris.sepallength < 5][:5])

Le résultat suivant s'affiche :

   sepallength  sepalwidth  petallength  petalwidth         name
0          4.9         3.0          1.4         0.2  Iris-setosa
1          4.7         3.2          1.3         0.2  Iris-setosa
2          4.6         3.1          1.5         0.2  Iris-setosa
3          4.6         3.4          1.4         0.3  Iris-setosa
4          4.4         2.9          1.4         0.2  Iris-setosa

Pour empêcher le système d'appeler automatiquement la méthode execute dans un environnement interactif, effectuez les opérations manuelles suivantes. L'exemple de code ci-dessous illustre cette procédure :

from odps import options
options.interactive = False

print(iris[iris.sepallength < 5][:5])

Le résultat suivant s'affiche :

Collection: ref_0
  odps.Table
    name: hudi_mc_0612.`iris3`
    schema:
      sepallength           : double      # Sepal length (cm)
      sepalwidth            : double      # Sepal width (cm)
      petallength           : double      # Petal length (cm)
      petalwidth            : double      # Petal width (cm)
      name                  : string      # Type
Collection: ref_1
  Filter[collection]
    collection: ref_0
    predicate:
      Less[sequence(boolean)]
        sepallength = Column[sequence(float64)] 'sepallength' from collection ref_0
        Scalar[int8]
          5
Slice[collection]
  collection: ref_1
  stop:
    Scalar[int8]
      5

Après avoir désactivé les appels automatiques, l'intégralité de l'arbre syntaxique abstrait (AST) s'affiche lors de l'appel à repr. Dans ce cas, vous devez appeler manuellement la méthode execute si vous souhaitez obtenir les résultats.

Récupérer les résultats d'exécution

Si un ResultFrame est renvoyé après l'appel de la méthode execute ou head, vous pouvez extraire les résultats depuis le ResultFrame.

Remarque

Un ResultFrame est un ensemble de résultats et ne peut pas être utilisé dans des calculs ultérieurs.

Parcourez itérativement tous les enregistrements du ResultFrame. L'exemple de code ci-dessous illustre cette procédure :

result = iris.head(3)
for r in result:
    print(list(r))

Le résultat suivant s'affiche :

[4.9, 3.0, 1.4, 0.2, 'Iris-setosa']
[4.7, 3.2, 1.3, 0.2, 'Iris-setosa']
[4.6, 3.1, 1.5, 0.2, 'Iris-setosa']

Si pandas est installé, convertissez un ResultFrame en pandas DataFrame ou en PyODPS DataFrame utilisant le backend pandas.

# Return a pandas DataFrame. 
pd_df = iris.head(3).to_pandas()

# Return a PyODPS DataFrame that uses the pandas backend. 
wrapped_df = iris.head(3).to_pandas(wrap=True)

Enregistrer les résultats dans des tables MaxCompute

Appelez la méthode persist pour renvoyer un nouvel objet DataFrame pour un objet Collection. La méthode persist utilise le nom de la table comme paramètre.

iris2 = iris[iris.sepalwidth < 2.5].persist('pyodps_iris')
print(iris2.head(5))

Le résultat suivant s'affiche :

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor

Spécifiez le paramètre partitions dans la méthode persist pour créer une table partitionnée. La table est partitionnée selon les colonnes spécifiées par partitions.

iris3 = iris[iris.sepalwidth < 2.5].persist('pyodps_iris_test', partitions=['name'])
print(iris3.data)

Le résultat suivant s'affiche :

odps.Table
  name: odps_test_sqltask_finance.`pyodps_iris`
  schema:
    sepallength           : double
    sepalwidth            : double
    petallength           : double
    petalwidth            : double
  partitions:
    name                  : string

Pour écrire des données dans une partition d'une table existante, spécifiez le paramètre partition dans la méthode persist. Le paramètre partition indique la partition cible. Par exemple, définissez le paramètre partition sur ds=******. La table doit contenir toutes les colonnes de l'objet DataFrame avec les mêmes types. Les paramètres drop_partition et create_partition ne sont valides que si le paramètre partition est spécifié. Le paramètre drop_partition indique s'il faut supprimer la partition spécifiée si elle existe. Le paramètre create_partition indique s'il faut créer la partition spécifiée si elle n'existe pas.
```
print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris_partition', partition='ds=test', drop_partition=True, create_partition=True).head(5))
```
Le résultat suivant s'affiche :
```
   sepallength  sepalwidth  petallength  petalwidth             name    ds
0          4.5         2.3          1.3         0.3      Iris-setosa  test
1          5.5         2.3          4.0         1.3  Iris-versicolor  test
2          4.9         2.4          3.3         1.0  Iris-versicolor  test
3          5.0         2.0          3.5         1.0  Iris-versicolor  test
4          6.0         2.2          4.0         1.0  Iris-versicolor  test
```

Lors de l'écriture des données dans une table, vous pouvez spécifier la durée de vie (TTL) de la table. Par exemple, l'instruction suivante définit la TTL de la table sur 10 jours.

print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris', lifecycle=10).head(5))

Le résultat suivant s'affiche :

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor

Si la source de données ne contient pas d'objets MaxCompute mais uniquement des objets pandas, spécifiez manuellement l'objet d'entrée MaxCompute ou marquez-le comme objet global lors de l'appel à la méthode persist.
```
# The entrance object is o. 
# Specify the entrance object. 
df.persist('table_name', odps=o)
# Alternative operation: Mark the entrance object as a global object. 
o.to_global()
df.persist('table_name')
```

Enregistrer les résultats dans un pandas DataFrame

Appelez la méthode to_pandas pour enregistrer les résultats dans un objet pandas DataFrame. Si le paramètre wrap est défini sur True, un objet PyODPS DataFrame est renvoyé.

Exemple 1 : Appelez la méthode to_pandas pour renvoyer un objet pandas DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas()))
```
Le résultat suivant s'affiche :
```
<class 'pandas.core.frame.DataFrame'>
```
Exemple 2 : Définissez le paramètre wrap sur True pour renvoyer un objet PyODPS DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas(wrap=True)))
```
Le résultat suivant s'affiche :
```
<class 'odps.df.core.DataFrame'>
```

Remarque

Vous pouvez appeler la méthode open_reader dans PyODPS et utiliser reader.to_pandas() pour convertir les résultats en objet pandas DataFrame. Pour plus d'informations, consultez la section Tables.

Configurer les paramètres d'exécution

Configurez les paramètres d'exécution pour les méthodes exécutées immédiatement, telles que execute, persist et to_pandas. Ce paramétrage n'est valide que pour le backend MaxCompute SQL.

Configurez les paramètres globaux. Pour plus d'informations, consultez la section SQL.

Spécifiez le paramètre hints dans ces méthodes. Cela garantit que les paramètres d'exécution spécifiés ne s'appliquent qu'au calcul en cours.

print(iris[iris.sepallength < 5].to_pandas(hints={'odps.sql.mapper.split.size': 16}))

Le résultat suivant s'affiche :

   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          4.9         2.4          3.3         1.0  Iris-versicolor

Afficher les détails au moment de l'exécution

Pour afficher les informations LogView d'une instance au moment de l'exécution, modifiez les configurations globales. L'exemple de code ci-dessous illustre cette procédure :

from odps import options
options.verbose = True

print(iris[iris.sepallength < 5].exclude('sepallength')[:5].execute())

Le résultat suivant s'affiche :

Sql compiled:
SELECT t1.`sepalwidth`, t1.`petallength`, t1.`petalwidth`, t1.`name`
FROM odps_test_sqltask_finance.`pyodps_iris` t1
WHERE t1.`sepallength` < 5
LIMIT 5
Instance ID:
  Log view:http://logview
  
   sepalwidth  petallength  petalwidth             name
0         2.3          1.3         0.3      Iris-setosa
1         2.4          3.3         1.0  Iris-versicolor

Spécifiez une fonction de journalisation. L'exemple de code ci-dessous illustre cette procédure :

my_logs = []
def my_logger(x):
    my_logs.append(x)
options.verbose_log = my_logger
print(iris[iris.sepallength < 5].exclude('sepallength')[:5].execute())

print(my_logs)

Le résultat suivant s'affiche :

   sepalwidth  petallength  petalwidth             name
0         2.3          1.3         0.3      Iris-setosa
1         2.4          3.3         1.0  Iris-versicolor

['Sql compiled:', 'CREATE TABLE tmp_pyodps_24332bdb_4fd0_4d0d_aed4_38a443618268 LIFECYCLE 1 AS \nSELECT t1.`sepalwidth`, t1.`petallength`, t1.`petalwidth`, t1.`name` \nFROM odps_test_sqltask_finance.`pyodps_iris` t1 \nWHERE t1.`sepallength` < 5 \nLIMIT 5', 'Instance ID: 20230815034706122gbymevg*****', '  Log view:]

Mettre en cache les résultats de calcul intermédiaires des objets Collection

Au cours du processus de calcul DataFrame, certains objets Collection sont utilisés plusieurs fois. Pour afficher les résultats d'exécution d'un processus intermédiaire, appelez la méthode cache pour marquer un objet Collection à calculer en priorité. L'exemple de code ci-dessous illustre cette procédure.

Remarque

L'exécution de la méthode cache est différée. L'appel à la méthode cache ne déclenche pas immédiatement le calcul automatique.

cached = iris[iris.sepalwidth < 3.5]['sepallength', 'name'].cache()
df = cached.head(3)
print(df)

# The following result is returned:
   sepallength             name
0          4.5      Iris-setosa
1          5.5  Iris-versicolor
2          4.9  Iris-versicolor

# You can immediately retrieve the calculation result because cached is calculated. 
print(cached.head(3))

# The following result is returned:
   sepallength             name
0          4.5      Iris-setosa
1          5.5  Iris-versicolor
2          4.9  Iris-versicolor

Exécution asynchrone et parallèle

Exécution asynchrone

PyODPS DataFrame prend en charge l'exécution asynchrone. Spécifiez le paramètre async pour activer l'exécution asynchrone pour les méthodes exécutées immédiatement suivantes : execute, persist, head, tail et to_pandas. Le paramètre timeout spécifie le délai d'expiration. Les opérations asynchrones renvoient des objets Future.

future = iris[iris.sepalwidth < 10].head(10, async_=True)
print(future.result())

# The following result is returned:
   sepallength  sepalwidth  petallength  petalwidth             name
0          4.5         2.3          1.3         0.3      Iris-setosa
1          5.5         2.3          4.0         1.3  Iris-versicolor
2          4.9         2.4          3.3         1.0  Iris-versicolor
3          5.0         2.0          3.5         1.0  Iris-versicolor
4          6.0         2.2          4.0         1.0  Iris-versicolor
5          6.2         2.2          4.5         1.5  Iris-versicolor
6          5.5         2.4          3.8         1.1  Iris-versicolor
7          5.5         2.4          3.7         1.0  Iris-versicolor
8          6.3         2.3          4.4         1.3  Iris-versicolor
9          5.0         2.3          3.3         1.0  Iris-versicolor

Exécution parallèle

Appelez l'opération Delay API récemment introduite pour différer les méthodes exécutées immédiatement suivantes : execute, persist, head, tail et to_pandas. Des objets Future sont alors renvoyés. Lors de l'appel à l'opération Delay API, le système identifie les dépendances et exécute les méthodes selon le niveau de concurrence spécifié. Dans ce cas, l'exécution asynchrone est prise en charge.

from odps.df import Delay
delay = Delay()  # Create a Delay object. 

df = iris[iris.sepal_width < 5].cache()  # Common dependency of subsequent expressions. 
future1 = df.sepal_width.sum().execute(delay=delay)  # Return a Future object. The execution is not started. 
future2 = df.sepal_width.mean().execute(delay=delay)
future3 = df.sepal_length.max().execute(delay=delay)
delay.execute(n_parallel=3)  # The execution starts with three concurrent threads. 
|==========================================|   1 /  1  (100.00%)        21s
print(future1.result())

# The following result is returned:
25.0
 
print(future2.result())

# The following result is returned:
2.272727272727273

Dans l'exemple précédent, PyODPS DataFrame exécute d'abord l'objet de la dépendance partagée. Ensuite, PyODPS DataFrame définit la concurrence sur 3 et exécute les objets de future1 à future3.

Spécifiez le paramètre async dans delay.execute pour indiquer s'il faut activer l'exécution asynchrone. Si l'exécution asynchrone est activée, vous pouvez également utiliser le paramètre timeout pour spécifier le délai d'expiration.