MaxCompute:PyODPS Sıra ve Çalıştırma İşlemleri

Yordam

1. Bir MaxCompute projesi oluşturduğunuzdan emin olun MaxCompute projesi oluşturma.

2. Bir DataWorks çalışma alanı oluşturduğunuzdan emin olun DataWorks çalışma alanı oluşturma. Bu konuda örnek olarak genel önizlemedeki bir DataStudio çalışma alanı kullanılır.

3. DataWorks'te pyodps_iris adında bir tablo oluşturun.

   1. DataWorks konsoluna giriş yapın ve sol üst köşeden bir bölge seçin.

   2. Workspaces sayfasında, hedef çalışma alanının Actions sütununda Shortcuts > DataStudio seçeneklerini seçin.

   3. Run Configuration sayfasında Computing Resource ve Resource Group seçeneklerini belirleyin.

      Kaynak grubunun oluşturulması birkaç dakika sürebilir. Ardından, Resource Groups sayfasında kaynak grubunu çalışma alanınıza bağlayın.

   4. Bir MaxCompute SQL düğümünde aşağıdaki ifadeyi çalıştırarak pyodps_iris tablosunu oluşturun.

      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. Test veri kümesini indirin ve MaxCompute'e aktarın.

   1. Iris flower veri kümesini indirip arşivden çıkarın ve ardından iris.data dosyasının adını iris.csv olarak değiştirin.

   2. DataWorks konsoluna giriş yapın ve sol üst köşeden bir bölge seçin.

   3. Sol gezinme bölmesinde Data Integration > Data Upload and Download seçeneklerini seçin.

   4. Go to Data Upload and Download seçeneğine tıklayın.

   5. Sol gezinme çubuğunda yükleme simgesine tıklayın ve Data Upload seçeneğini seçin.

5. DataStudio'da bir MaxCompute PyODPS 2 düğümü oluşturun. Aşağıdaki örnek kodu girin ve Çalıştır'a tıklayın.

   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 adlı bir PyODPS düğümü oluşturun ve aşağıdaki kodu çalıştırın:

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