MaxCompute:Sekuens PyODPS dan operasi eksekusi

Prosedur

1. Pastikan Anda telah membuat proyek MaxCompute.

2. Pastikan Anda telah membuat ruang kerja DataWorks. Topik ini menggunakan ruang kerja DataStudio dalam pratinjau publik sebagai contoh.

3. Di DataWorks, buat tabel pyodps_iris.

   1. Masuk ke Konsol DataWorks dan pilih Wilayah di pojok kiri atas.

   2. Pada halaman Workspaces, pada kolom Actions dari ruang kerja target, pilih Shortcuts > DataStudio.

   3. Pada halaman Run Configuration, pilih Computing Resource dan Resource Group.

      Kelompok sumber daya mungkin memerlukan beberapa menit untuk dibuat. Setelah itu, pada halaman Resource Groups, bind kelompok sumber daya tersebut ke ruang kerja Anda.

   4. Pada node SQL MaxCompute, jalankan pernyataan berikut untuk membuat tabel 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. Unduh set data uji dan impor ke MaxCompute.

   1. Unduh dan ekstrak dataset Iris flower, lalu ubah nama file iris.data menjadi iris.csv.

   2. Masuk ke Konsol DataWorks dan pilih Wilayah di pojok kiri atas.

   3. Pada panel navigasi kiri, pilih Data Integration > Data Upload and Download.

   4. Klik Go to Data Upload and Download.

   5. Pada bilah navigasi kiri, klik ikon unggah , lalu klik Data Upload.

5. Di DataStudio, buat node MaxCompute PyODPS 2. Masukkan kode contoh berikut dan klik Run.

   from odps import DataFrame
   iris = DataFrame(o.get_table('pyodps_iris'))
   
   # Dapatkan sebuah kolom.
   print iris.sepallength.head(5)
   
   print iris['sepallength'].head(5)
   
   # Lihat tipe data kolom tersebut.
   print iris.sepallength.dtype
   
   # Ubah tipe data kolom tersebut.
   iris.sepallength.astype('int')
   
   # Lakukan komputasi.
   print iris.groupby('name').sepallength.max().head(5)
   
   print iris.sepallength.max()
   
   # Ubah nama kolom.
   print iris.sepalwidth.rename('speal_width').head(5)
   
   # Transformasi kolom sederhana.
   print (iris.sepallength + iris.sepalwidth).rename('sum_sepal').head(5)

6. Buat node PyODPS bernama PyExecute dan jalankan kode berikut:

   from odps import options
   from odps import DataFrame
   
   # Lihat Logview dari instans yang sedang berjalan.
   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 hasil Collection antara.
   cached = iris[iris.sepalwidth < 3.5].cache()
   print cached.head(3)
   
   # Eksekusi paralel asinkron.
   from odps.df import Delay
   delay = Delay() # Buat objek Delay.
   df = iris[iris.sepalwidth < 5].cache()  # Ini membuat dependensi bersama.
   future1 = df.sepalwidth.sum().execute(delay=delay) # Langsung mengembalikan objek future tanpa memulai eksekusi.
   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()