MaxCompute:PyODPS シーケンスと実行操作

操作手順

1. MaxCompute プロジェクトを作成済みであることを確認します。

2. DataWorks ワークスペースを作成済みであることを確認します。本トピックでは、パブリックプレビュー中の DataStudio ワークスペースを例として使用します。

3. DataWorks で pyodps_iris テーブルを作成します。

   1. DataWorks コンソールにログインし、左上隅でリージョンを選択します。

   2. ワークスペース一覧ページで、対象のワークスペースの 操作 列から、ショートカット > [DataStudio] を選択します。

   3. デバッグ設定ページで、計算リソースおよび リソースグループを選択します。

      リソースグループの作成には数分かかる場合があります。その後、リソースグループの一覧ページで、リソースグループをご利用のワークスペースにバインドします。

   4. MaxCompute SQL ノードで、次の文を実行して 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. テストデータセットをダウンロードし、MaxCompute にインポートします。

   1. Iris flower データセットをダウンロードして解凍し、iris.data ファイルを iris.csv に名前変更します。

   2. DataWorks コンソールにログインし、左上隅でリージョンを選択します。

   3. 左側のナビゲーションウィンドウで、Data Integration > データのアップロードとダウンロード を選択します。

   4. ［移動］［データのアップロードとダウンロード］をクリックします。

   5. 左側のナビゲーションバーでアップロードアイコン をクリックし、データのアップロード をクリックします。

5. DataStudio で MaxCompute [PyODPS 2] ノードを作成します。次のサンプルコードを入力し、実行します。

   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. PyExecute という名前の PyODPS ノードを作成し、次のコードを実行します。

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