MaxCompute:PyODPS Sequence and execution operations

Procedure

1. Ensure that you have created a MaxCompute project.

2. Ensure that you have created a DataWorks workspace. This topic uses a DataStudio workspace in public preview as an example.

3. In DataWorks, create a pyodps_iris table.

   1. Log in to the DataWorks console and select a region in the upper-left corner.

   2. On the Workspaces page, in the Actions column of the target workspace, choose Shortcuts > DataStudio.

   3. On the Run Configuration page, select Computing Resource and Resource Group.

      The resource group may take a few minutes to create. Then, on the Resource Groups page, bind the resource group to your workspace.

   4. In a MaxCompute SQL node, run the following statement to create the pyodps_iris 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'
      );

4. Download the test dataset and import it into MaxCompute.

   1. Download and decompress the Iris flower dataset, and then rename the iris.data file to iris.csv.

   2. Log in to the DataWorks console and select a region in the upper-left corner.

   3. In the navigation pane on the left, choose Data Integration > Data Upload and Download.

   4. Click Go to Data Upload and Download.

   5. In the left navigation bar, click the upload icon , and click Data Upload.

5. In DataStudio, create a MaxCompute PyODPS 2 node. Enter the following sample code and click 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. Create a PyODPS node named PyExecute and run the following code:

   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()