Execute and obtain results - MaxCompute - Centro de documentación de Alibaba Cloud

Este tema describe los métodos de ejecución disponibles para las operaciones de DataFrame.

Requisitos previos

Asegúrese de cumplir con los siguientes requisitos:

Debe contar con una tabla de ejemplo llamada pyodps_iris. Para más información, consulte DataFrame data processing.
Debe haber creado un objeto DataFrame. Para más información, consulte la sección "Crear un objeto DataFrame a partir de una tabla de MaxCompute" en Create a DataFrame object.

Ejecución diferida

Las operaciones de DataFrame se ejecutan únicamente cuando se invoca explícitamente el método execute o cuando se llaman métodos que internamente invocan a execute. La siguiente tabla enumera los métodos que llaman internamente al método execute.

Método	Descripción	Valor de retorno
persist	Guarda los resultados de la ejecución en tablas de MaxCompute.	PyODPS DataFrame
execute	Ejecuta las operaciones y devuelve todos los resultados.	ResultFrame
head	Ejecuta las operaciones y devuelve las primeras N filas de datos resultantes.	ResultFrame
tail	Ejecuta las operaciones y devuelve las últimas N filas de datos resultantes.	ResultFrame
to_pandas	Convierte un objeto Collection en un objeto pandas DataFrame, o bien convierte un objeto Sequence en un objeto Series. Si el parámetro wrap se establece en True, se devuelve un objeto PyODPS DataFrame.	Si el parámetro wrap se establece en True, se devuelve un objeto PyODPS DataFrame. Si el parámetro wrap se establece en False, se devuelve un objeto pandas DataFrame. El valor predeterminado del parámetro wrap es False.
plot, hist y boxplot	Métodos de graficación.	N/A

Nota

En un entorno interactivo, PyODPS DataFrame llama automáticamente al método execute cuando muestra datos de resultados o cuando invoca el método repr. No es necesario llamar manualmente al método execute.

Ejemplos

# 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])

Se obtiene el siguiente resultado:

   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

Si necesita evitar que el sistema llame automáticamente al método execute en un entorno interactivo, debe realizar una configuración manual. El siguiente código muestra un ejemplo:

from odps import options
options.interactive = False

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

Se obtiene el siguiente resultado:

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

Al deshabilitar las llamadas automáticas, se muestra el árbol de sintaxis abstracta (AST) completo cuando se visualiza el objeto repr. En este caso, debe llamar manualmente al método execute si requiere utilizarlo.

Obtener resultados de ejecución

Si tras invocar el método execute o head se devuelve un objeto ResultFrame, puede recuperar los resultados desde dicho ResultFrame.

Nota

Un ResultFrame es un conjunto de resultados y no puede utilizarse en cálculos posteriores.

Puede recuperar iterativamente todos los registros del ResultFrame. El siguiente código muestra un ejemplo:

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

Se obtiene el siguiente resultado:

[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á instalado, un ResultFrame puede convertirse en un pandas DataFrame o en un PyODPS DataFrame que utilice el backend de 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)

Guardar resultados en tablas de MaxCompute

Puede llamar al método persist para devolver un nuevo objeto DataFrame correspondiente a un objeto Collection. El método persist utiliza el nombre de la tabla como parámetro.

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

Se obtiene el siguiente resultado:

   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

Puede especificar el parámetro partitions en el método persist para crear una tabla particionada. La tabla se particiona según las columnas indicadas por partitions.

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

Se obtiene el siguiente resultado:

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

Para escribir datos en una partición de una tabla existente, especifique el parámetro partition en el método persist. Este parámetro indica la partición destino de la escritura. Por ejemplo, establezca el parámetro partition como ds=******. La tabla debe contener todas las columnas del objeto DataFrame y estas deben ser del mismo tipo. Los parámetros drop_partition y create_partition solo son válidos si se especifica el parámetro partition. El parámetro drop_partition determina si se elimina la partición indicada en caso de que ya exista. El parámetro create_partition determina si se crea la partición indicada en caso de que no exista.
```
print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris_partition', partition='ds=test', drop_partition=True, create_partition=True).head(5))
```
Se obtiene el siguiente resultado:
```
   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
```

Al escribir datos en una tabla, puede definir su tiempo de vida (TTL). Por ejemplo, la siguiente instrucción establece el TTL de la tabla en 10 días.

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

Se obtiene el siguiente resultado:

   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 el origen de datos no contiene objetos de MaxCompute y solo incluye objetos de pandas, debe especificar manualmente el objeto de entrada de MaxCompute o marcarlo como objeto global al invocar el método 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')
```

Guardar resultados en un pandas DataFrame

Puede llamar al método to_pandas para guardar los resultados en un objeto pandas DataFrame. Si el parámetro wrap se establece en True, se devuelve un objeto PyODPS DataFrame.

Ejemplo 1: Llame al método to_pandas para devolver un objeto pandas DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas()))
```
Se obtiene el siguiente resultado:
```
<class 'pandas.core.frame.DataFrame'>
```
Ejemplo 2: Establezca el parámetro wrap en True para devolver un objeto PyODPS DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas(wrap=True)))
```
Se obtiene el siguiente resultado:
```
<class 'odps.df.core.DataFrame'>
```

Nota

Puede llamar al método open_reader en PyODPS y usar reader.to_pandas() para convertir los resultados en un objeto pandas DataFrame. Para más información, consulte Tables.

Configurar parámetros de tiempo de ejecución

Es posible configurar los parámetros de tiempo de ejecución para métodos de ejecución inmediata, tales como execute, persist y to_pandas. Esta configuración solo aplica al backend de MaxCompute SQL.

Configure los parámetros globales. Para más información, consulte SQL.

Especifique el parámetro hints en estos métodos. Esto garantiza que los parámetros de tiempo de ejecución indicados sean válidos únicamente para el cálculo actual.

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

Se obtiene el siguiente resultado:

   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

Mostrar detalles durante la ejecución

Para consultar la información de LogView de una instancia durante la ejecución, debe modificar las configuraciones globales. El siguiente código muestra un ejemplo:

from odps import options
options.verbose = True

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

Se obtiene el siguiente resultado:

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

Puede especificar una función de registro. El siguiente código muestra un ejemplo:

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)

Se obtiene el siguiente resultado:

   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:]

Almacenar en caché resultados intermedios de objetos Collection

Durante el proceso de cálculo de DataFrame, algunos objetos Collection se utilizan varias veces. Para ver los resultados de ejecución de un proceso intermedio, puede llamar al método cache para marcar un objeto Collection que desea calcular primero. El siguiente código muestra un ejemplo.

Nota

La ejecución del método cache es diferida. No se desencadena un cálculo automático inmediatamente después de invocar el método cache.

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

Ejecución asíncrona y paralela

Ejecución asíncrona

PyODPS DataFrame admite la ejecución asíncrona. Puede especificar el parámetro async para habilitar la ejecución asíncrona en los siguientes métodos de ejecución inmediata: execute, persist, head, tail y to_pandas. El parámetro timeout define el periodo de tiempo de espera. Las operaciones asíncronas devuelven objetos 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

Ejecución paralela

Puede utilizar la nueva operación Delay API para diferir los siguientes métodos de ejecución inmediata: execute, persist, head, tail y to_pandas. Posteriormente, se devuelven objetos Future. Cuando se invoca la operación Delay API, el sistema identifica las dependencias y ejecuta los métodos según la concurrencia especificada. En este escenario también se admite la ejecución asíncrona.

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

En el ejemplo anterior, PyODPS DataFrame ejecuta primero el objeto de la dependencia compartida. Después, establece la concurrencia en 3 y ejecuta los objetos desde future1 hasta future3.

Puede especificar el parámetro async en delay.execute para indicar si se habilita la ejecución asíncrona. Si esta se encuentra habilitada, también puede usar el parámetro timeout para definir el periodo de tiempo de espera.