MaxCompute:PyODPS

Skenario umum penggunaan PyODPS:

• Memulai Cepat dengan DataFrame.

• Membaca data dari tabel partisi menggunakan PyODPS.

• Meneruskan parameter dalam PyODPS.

• Menggunakan paket pihak ketiga dalam PyODPS.

• Melihat partisi hash dalam PyODPS.

• Melakukan kueri bersyarat dalam PyODPS.

• Operasi sekuens dan eksekusi dalam PyODPS.

PyODPS dapat dijalankan di lingkungan lokal, DataWorks, dan Notebook PAI.

• Lingkungan lokal: Instal dan gunakan PyODPS secara lokal. Menggunakan PyODPS di lingkungan lokal.

• DataWorks: PyODPS telah dipra-instal pada node PyODPS DataWorks untuk mengembangkan dan menjadwalkan tugas. Menggunakan PyODPS dengan DataWorks.

• Notebook PAI: Anda dapat menginstal dan menjalankan PyODPS di lingkungan Python PAI. PyODPS juga telah dipra-instal dalam image bawaan PAI, termasuk komponen Python kustom PAI-Designer. Penggunaannya mengikuti metode standar. Operasi dasar dan DataFrame (Tidak disarankan).

PyODPS berjalan di berbagai klien, termasuk PC, node PyODPS DataWorks, dan lingkungan Notebook PAI. PyODPS menyediakan operasi seperti unduh melalui saluran data (tunnel download), execute, dan to_pandas yang menarik data ke mesin lokal. Pengguna baru sering kali menarik data secara lokal, memprosesnya, lalu mengunggahnya kembali — pendekatan ini tidak efisien karena melewati kemampuan komputasi terdistribusi MaxCompute.

Contoh tokenisasi berikut membandingkan kedua metode tersebut.

• Contoh skenario

  Seorang pengguna menganalisis string log harian dari tabel string satu kolom. Tujuannya adalah melakukan tokenisasi kalimat bahasa Tionghoa dengan pustaka jieba, menemukan kata kunci, lalu menyimpan hasilnya ke tabel baru.

• Demo kode pemrosesan tidak efisien

  import jieba
  t = o.get_table('word_split')
  out = []
  with t.open_reader() as reader:
      for r in reader:
          words = list(jieba.cut(r[0]))
          #
          # Logika pemrosesan untuk menghasilkan processed_data
          #
          out.append(processed_data)
  out_t = o.get_table('words')
  with out_t.open_writer() as writer:
      writer.write(out)

  Pendekatan single-machine ini membaca, memproses, dan menulis data baris per baris. Siklus unduh-unggah lambat dan memakan banyak memori. Di node DataWorks, pekerjaan sering mengalami error OOM karena memori default yang terbatas.

• Demo kode pemrosesan efisien

  from odps.df import output
  out_table = o.get_table('words')
  df = o.get_table('word_split').to_df()
  
  # Asumsikan bidang dan tipe yang dikembalikan adalah sebagai berikut
  out_names = ["word", "count"]
  out_types = ["string", "int"]
  
  @output(out_names, out_types)
  def handle(row):
      import jieba
      words = list(jieba.cut(row[0]))
      #
      # Logika pemrosesan untuk menghasilkan processed_data
      #
      yield processed_data
  df.apply(handle, axis=1).persist(out_table.name)

  Gunakan apply untuk menerapkan eksekusi terdistribusi:

  • Fungsi handle diserialisasi dan dieksekusi di sisi server sebagai UDF. Logika inti pemrosesan baris per baris tetap sama, tetapi MaxCompute mengeksekusinya secara paralel di beberapa mesin.

  • Antarmuka persist menulis output langsung ke tabel MaxCompute, sehingga semua data tetap berada di dalam kluster dan menghemat bandwidth serta memori lokal.

  • MaxCompute mendukung paket pihak ketiga seperti jieba dalam UDF, sehingga Anda dapat memanfaatkan komputasi terdistribusi dengan perubahan kode yang minimal.