MaxCompute:Séquences et opérations d'exécution PyODPS

Procédure

1. Assurez-vous d'avoir créé un projet MaxCompute.

2. Assurez-vous d'avoir créé un espace de travail DataWorks. Cette rubrique utilise un espace de travail DataStudio en aperçu public à titre d'exemple.

3. Dans DataWorks, créez une table pyodps_iris.

   1. Connectez-vous à la console DataWorks et sélectionnez une région dans le coin supérieur gauche.

   2. Sur la page Workspaces, dans la colonne Actions de l'espace de travail cible, choisissez Shortcuts > DataStudio.

   3. Sur la page Run Configuration, sélectionnez Computing Resource et Resource Group.

      La création du groupe de ressources peut prendre quelques minutes. Ensuite, sur la page Resource Groups, liez le groupe de ressources à votre espace de travail.

   4. Dans un nœud SQL MaxCompute, exécutez l'instruction suivante pour créer la table 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. Téléchargez le jeu de données de test et importez-le dans MaxCompute.

   1. Téléchargez et décompressez le jeu de données Iris flower, puis renommez le fichier iris.data en iris.csv.

   2. Connectez-vous à la console DataWorks et sélectionnez une région dans le coin supérieur gauche.

   3. Dans le volet de navigation de gauche, choisissez Data Integration > Data Upload and Download.

   4. Cliquez sur Go to Data Upload and Download.

   5. Dans la barre de navigation de gauche, cliquez sur l'icône de téléchargement , puis cliquez sur Data Upload.

5. Dans DataStudio, créez un nœud PyODPS 2 MaxCompute. Saisissez l'exemple de code suivant et cliquez sur 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. Créez un nœud PyODPS nommé PyExecute et exécutez le code suivant :

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