MaxCompute:Başlarken

Önkoşullar

• MaxCompute hizmetini etkinleştirdiniz.

• Bir MaxCompute projesi oluşturdunuz.

• Bir DataWorks çalışma alanı oluşturdunuz ve buna hesaplama kaynaklarını bağladınız.

Yordam

1. Bir PyODPS düğümü oluşturun.

   Bu konuda bir DataWorks PyODPS düğümü kullanıyoruz. Ayrıntılar için bkz. PyODPS 3 görevi geliştirme.

   • Örnek olarak PyODPS 3 düğümü kullanıyoruz. PyODPS 3 düğümlerinin temel Python sürümü 3,7'dir.

   • Bir PyODPS düğümü en fazla 50 MB yerel veriyi işleyebilir ve çalışma zamanında 1 GB'a kadar bellek kullanabilir. Bir görev bu sınırları aşarsa sistem görevi sonlandırır. Bu nedenle PyODPS görevinde büyük miktarda veri işleyen Python kodu yazmaktan kaçının.

   • DataWorks'te kod yazma ve hata ayıklama daha az verimlidir. Verimliliği artırmak için kodunuzu IntelliJ IDEA'da yerel ortamda geliştirmenizi öneririz.

   1. Bir iş akışı oluşturun.

      Data Studio sayfasına gidin Business Flow öğesine sağ tıklayın ve ardından Create Workflow seçeneğini seçin.

   2. Bir PyODPS düğümü oluşturun.

      Yeni iş akışına sağ tıklayın Create Node > MaxCompute > PyODPS 3 yolunu izleyin, düğüm için bir ad girin ve ardından Submit öğesini tıklayın.

2. PyODPS düğümünü düzenleyin.

   1. Kodu yazın.

      Test kodunu PyODPS düğümü düzenleyicisine girin. Aşağıdaki örnek tablo işlemleri için PyODPS API'lerinin kullanımını gösterir. Tablo işlemleri ve SQL işlemleri hakkında daha fazla bilgi için bkz. Tablolar ve SQL.

      from odps import ODPS
      
      # In DataWorks PyODPS nodes, a global variable (`o` or `odps`) is available by default as the MaxCompute entry. You can use it directly without defining it manually.
      table = o.create_table('my_new_table', 'num bigint, id string', if_not_exists=True)
      
      # Insert data into the non-partitioned table my_new_table.
      records = [[111, 'aaa'],
                [222, 'bbb'],
                [333, 'ccc'],
                [444, '中文']]
      o.write_table(table, records)
      
      # Read data from the non-partitioned table my_new_table.
      for record in o.read_table(table):
          print(record[0],record[1])
      
      # Read data from the table by executing an SQL statement.
      result = o.execute_sql('select * from my_new_table;',hints={'odps.sql.allow.fullscan': 'true'})
      
      # Read the SQL execution results.
      with result.open_reader() as reader:    
          for record in reader:            
              print(record[0],record[1])
      
      # Drop the table to release resources.
      table.drop()

   2. Kodu çalıştırın.

      Kodu düzenledikten sonra simgesine tıklayın. Çalıştırma tamamlandığında sonuçları Runtime Log sekmesinde görüntüleyebilirsiniz. Aşağıdaki günlük çıktısı başarılı olduğunu gösterir.

      2023-07-21 15:06:41 INFO ========================================================
      Executing user script with PyODPS 0.11.2.3
      Tunnel session created: <TableUploadSession xxx                                          >
      Tunnel session created: <TableDownloadSession xxx                                        >
      111 aaa
      222 bbb
      333 ccc
      444 中文
      Tunnel session created: <InstanceDownloadSession id=xxx          project_name=xxx                              >
      111 aaa
      222 bbb
      333 ccc
      444 中文
      2023-07-21 15:06:49 INFO ========================================================
      2023-07-21 15:06:49 INFO Exit code of the Shell command 0
      2023-07-21 15:06:49 INFO --- Invocation of Shell command completed ---
      2023-07-21 15:06:49 INFO Shell run successfully!
      2023-07-21 15:06:49 INFO Current task status: FINISH
      2023-07-21 15:06:49 INFO Cost time is: 7.507s
      /home/admin/alisatasknode/taskinfo/xxx              xxx        .log-END-EOF