MaxCompute:Operações de sequência e execução no PyODPS

Procedimento

1. Verifique se você criou um projeto do MaxCompute.

2. Confirme se você criou um workspace do DataWorks. Este tópico usa como exemplo um workspace do DataStudio em preview público.

3. No DataWorks, crie uma tabela pyodps_iris.

   1. Faça login no console do DataWorks e selecione uma região no canto superior esquerdo.

   2. Na página Workspaces, na coluna Actions do workspace desejado, escolha Shortcuts > DataStudio.

   3. Na página Run Configuration, selecione Computing Resource e Resource Group.

      A criação do grupo de recursos pode levar alguns minutos. Depois, na página Resource Groups, vincule o grupo de recursos ao seu workspace.

   4. Em um nó MaxCompute SQL, execute a instrução a seguir para criar a tabela pyodps_iris.

      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'
      );

4. Baixe o conjunto de dados de teste e importe-o para o MaxCompute.

   1. Baixe e descompacte o conjunto de dados Iris flower e renomeie o arquivo iris.data para iris.csv.

   2. Acesse o console do DataWorks e selecione uma região no canto superior esquerdo.

   3. No painel de navegação à esquerda, escolha Data Integration > Data Upload and Download.

   4. Clique em Go to Data Upload and Download.

   5. Na barra de navegação à esquerda, clique no ícone de upload e clique em Data Upload.

5. No DataStudio, crie um nó MaxCompute PyODPS 2. Insira o código de exemplo a seguir e clique em Run.

   from odps import DataFrame
   iris = DataFrame(o.get_table('pyodps_iris'))
   
   # Get a column.
   print iris.sepallength.head(5)
   
   print iris['sepallength'].head(5)
   
   # View the data type of the column.
   print iris.sepallength.dtype
   
   # Change the data type of the column.
   iris.sepallength.astype('int')
   
   # Compute.
   print iris.groupby('name').sepallength.max().head(5)
   
   print iris.sepallength.max()
   
   # Rename a column.
   print iris.sepalwidth.rename('speal_width').head(5)
   
   # Simple column transformation.
   print (iris.sepallength + iris.sepalwidth).rename('sum_sepal').head(5)

6. Crie um nó PyODPS chamado PyExecute e execute o código a seguir:

   from odps import options
   from odps import DataFrame
   
   # View the Logview of the running instance.
   options.verbose = True
   iris = DataFrame(o.get_table('pyodps_iris'))
   iris[iris.sepallength < 5].exclude('sepallength')[:5].execute()
   
   my_logs = []
   def my_loggers(x):
     my_logs.append(x)
   
   options.verbose_log = my_loggers
   
   iris[iris.sepallength < 5].exclude('sepallength')[:5].execute()
   
   print(my_logs)
   
   # Cache an intermediate Collection result.
   cached = iris[iris.sepalwidth < 3.5].cache()
   print cached.head(3)
   
   # Asynchronous parallel execution.
   from odps.df import Delay
   delay = Delay() # Create a Delay object.
   df = iris[iris.sepalwidth < 5].cache()  # This creates a shared dependency.
   future1 = df.sepalwidth.sum().execute(delay=delay) # Immediately returns a future object without starting execution.
   future2 = df.sepalwidth.mean().execute(delay=delay)
   future3 = df.sepalwidth.max().execute(delay=delay)
   
   delay.execute(n_parallel=3)
   
   print future1.result()
   print future2.result()
   print future3.result()