Execute and obtain results - MaxCompute: Central de Documentação do Alibaba Cloud

Este tópico descreve os métodos de execução disponíveis para operações de DataFrame.

Pré-requisitos

Certifique-se de que os seguintes requisitos foram atendidos:

Uma tabela de amostra chamada pyodps_iris está preparada. Para mais informações, consulte DataFrame data processing.
Um objeto DataFrame foi criado. Para mais informações, consulte a seção "Criar um objeto DataFrame a partir de uma tabela do MaxCompute" em Create a DataFrame object.

Execução adiada

As operações de DataFrame são executadas apenas quando você chama explicitamente o método execute ou utiliza métodos que invocam internamente esse mesmo método. A tabela a seguir lista os métodos que chamam internamente o método execute.

Método	Descrição	Valor de retorno
persist	Salva os resultados da execução em tabelas do MaxCompute.	PyODPS DataFrame
execute	Executa as operações e retorna todos os resultados.	ResultFrame
head	Executa as operações e retorna as primeiras N linhas dos dados de resultado.	ResultFrame
tail	Executa as operações e retorna as últimas N linhas dos dados de resultado.	ResultFrame
to_pandas	Converte um objeto Collection em um objeto pandas DataFrame ou converte um objeto Sequence em um objeto Series. Se o parâmetro wrap for definido como True, um objeto PyODPS DataFrame será retornado.	Se o parâmetro wrap for definido como True, um objeto PyODPS DataFrame é retornado. Se o parâmetro wrap for definido como False, um objeto pandas DataFrame é retornado. O valor padrão do parâmetro wrap é False.
plot, hist e boxplot	Métodos de plotagem.	N/A

Nota

Em um ambiente interativo, o PyODPS DataFrame chama automaticamente o método execute ao exibir dados de resultado ou ao invocar o método repr. Não é necessário chamar manualmente o método execute.

Exemplos

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

O seguinte resultado é retornado:

   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

Para desativar a chamada automática do método execute em um ambiente interativo, realize as operações manuais. O código a seguir mostra um exemplo:

from odps import options
options.interactive = False

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

O seguinte resultado é retornado:

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

Após desativar as chamadas automáticas, toda a árvore de sintaxe abstrata (AST) é exibida quando o objeto repr é mostrado. Nesse caso, chame manualmente o método execute se precisar utilizá-lo.

Recuperar resultados da execução

Se um ResultFrame for retornado após a chamada do método execute ou head, recupere os resultados diretamente do ResultFrame.

Nota

Um ResultFrame é um conjunto de resultados e não pode ser usado em cálculos subsequentes.

Recupere iterativamente todos os registros do ResultFrame. O código a seguir mostra um exemplo:

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

O seguinte resultado é retornado:

[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']

Se o pandas estiver instalado, converta um ResultFrame em um pandas DataFrame ou em um PyODPS DataFrame que utilize o 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)

Salvar resultados em tabelas do MaxCompute

Chame o método persist para retornar um novo objeto DataFrame para um objeto Collection. O método persist utiliza o nome da tabela como parâmetro.

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

O seguinte resultado é retornado:

   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

Para criar uma tabela particionada, especifique o parâmetro partitions no método persist. A tabela será particionada com base nas colunas definidas por partitions.

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

O seguinte resultado é retornado:

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

Para gravar dados em uma partição de uma tabela existente, especifique o parâmetro partition no método persist. Esse parâmetro define a partição de destino da gravação. Por exemplo, defina o parâmetro partition como ds=******. A tabela deve conter todas as colunas do objeto DataFrame e essas colunas devem ser do mesmo tipo. Os parâmetros drop_partition e create_partition são válidos apenas se o parâmetro partition for especificado. O parâmetro drop_partition determina se a partição especificada deve ser excluída caso já exista. Já o parâmetro create_partition indica se a partição deve ser criada caso não exista.
```
print(iris[iris.sepalwidth < 2.5].persist('pyodps_iris_partition', partition='ds=test', drop_partition=True, create_partition=True).head(5))
```
O seguinte resultado é retornado:
```
   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
```

Ao gravar dados em uma tabela, defina o tempo de vida (TTL) da tabela. Por exemplo, a instrução a seguir define o TTL da tabela como 10 dias.

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

O seguinte resultado é retornado:

   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

Se a source de dados não contiver objetos do MaxCompute e tiver apenas objetos pandas, especifique manualmente o objeto de entrada do MaxCompute ou marque-o como objeto global ao chamar o 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')
```

Salvar resultados em pandas DataFrame

Utilize o método to_pandas para salvar resultados em um objeto pandas DataFrame. Se o parâmetro wrap estiver definido como True, um objeto PyODPS DataFrame será retornado.

Exemplo 1: Chame o método to_pandas para retornar um objeto pandas DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas()))
```
O seguinte resultado é retornado:
```
<class 'pandas.core.frame.DataFrame'>
```
Exemplo 2: Defina o parâmetro wrap como True para retornar um objeto PyODPS DataFrame.
```
print(type(iris[iris.sepalwidth < 2.5].to_pandas(wrap=True)))
```
O seguinte resultado é retornado:
```
<class 'odps.df.core.DataFrame'>
```

Nota

No PyODPS, chame o método open_reader e use reader.to_pandas() para converter os resultados em um objeto pandas DataFrame. Para mais informações, consulte Tables.

Configurar parâmetros de tempo de execução

Configure os parâmetros de tempo de execução para métodos de execução imediata, como execute, persist e to_pandas. Essa configuração é válida apenas para o backend MaxCompute SQL.

Configure os parâmetros globais. Para mais informações, consulte SQL.

Especifique o parâmetro hints nesses métodos. Isso garante que os parâmetros de tempo de execução especificados sejam válidos apenas para o cálculo atual.

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

O seguinte resultado é retornado:

   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

Exibir detalhes durante a execução

Para visualizar as informações de LogView de uma instância durante a execução, modifique as configurações globais. O código a seguir mostra um exemplo:

from odps import options
options.verbose = True

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

O seguinte resultado é retornado:

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

Defina uma função de logging personalizada. O código a seguir mostra um exemplo:

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)

O seguinte resultado é retornado:

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

Armazenar em cache resultados de cálculos intermediários de objetos Collection

Durante o processo de cálculo do DataFrame, alguns objetos Collection são utilizados várias vezes. Para visualizar os resultados de execução de um processo intermediário, chame o método cache para marcar um objeto Collection que deseja calcular primeiro. O código a seguir mostra um exemplo.

Nota

A execução do método cache é adiada. O cálculo automático não é acionado imediatamente após a chamada desse método.

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

Execução assíncrona e paralela

Execução assíncrona

O PyODPS DataFrame suporta execução assíncrona. Especifique o parâmetro async para ativar a execução assíncrona nos seguintes métodos de execução imediata: execute, persist, head, tail e to_pandas. O parâmetro timeout define o período de tempo limite. As operações assíncronas retornam 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

Execução paralela

Utilize a operação Delay API recém-introduzida para adiar os seguintes métodos de execução imediata: execute, persist, head, tail e to_pandas. Em seguida, objetos Future são retornados. Quando a operação Delay API é chamada, o sistema identifica as dependências e executa os métodos com base na concorrência especificada. Nesse cenário, há suporte à execução assí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

No exemplo anterior, o PyODPS DataFrame executa primeiro o objeto da dependência compartilhada. Depois, define a concorrência como 3 e executa os objetos de future1 até future3.

Especifique o parâmetro async em delay.execute para definir se a execução assíncrona deve ser ativada. Se a execução assíncrona estiver ativada, utilize também o parâmetro timeout para especificar o período de tempo limite.