MaxCompute:Praktik terbaik untuk menggunakan operator MaxFrame apply_chunk

Deskripsi parameter

Secara eksplisit deklarasikan output_type dan dtypes

• Jangan mengandalkan inferensi tipe karena dapat menyebabkan kegagalan waktu proses atau degradasi performa.

  result_df = df.mf.apply_chunk(<process>)  # dtypes tidak disertakan!

• Rekomendasi: pernyataan eksplisit

  Penting

  Jangan langsung memodifikasi dtypes dari df asli. Sebagai gantinya, gunakan .copy() atau buat DataFrame baru.

  Modifikasi kode sebelumnya sebagai berikut:

  def process(df_chunk):
      # Anda dapat menambahkan logika di sini atau cukup mengembalikan chunk tersebut. Kuncinya adalah struktur output.
      return df_chunk.copy() # Mengembalikan DataFrame yang berisi 'A' dan 'B'.
      
  def get_incorrect_dtypes(df):
      # Mengembalikan dtypes yang tidak lengkap yang sengaja tidak sesuai dengan output UDF.
      return df.dtypes.drop('A')
  
  incorrect_dtypes = get_incorrect_dtypes(df)
        
  print("\n--- Mencoba memanggil apply_chunk dengan dtypes yang tidak sesuai (error diharapkan) ---")
  try:
      result_df = df.mf.apply_chunk(
          process,
          output_type="dataframe",
          dtypes=incorrect_dtypes  # Menggunakan dtypes yang salah.
      ).execute()
  except Exception as e:
      print("Berhasil menangkap error yang diharapkan! Pesan error sebagai berikut:")
      print(f"Tipe error: {type(e)}")
      print(f"Detail error: {e}\n")
      # Error biasanya mirip dengan 'ValueError: forward method expected 1 arguments, but get 2'.
      # Hal ini karena dtypes hanya menentukan satu kolom ('B'), tetapi UDF mengembalikan dua kolom ('A', 'B').
  
  print("--- Penggunaan yang benar adalah sebagai berikut ---")
  #    Cara yang benar adalah dtypes harus secara akurat menggambarkan output UDF (process).
  #    Karena process mengembalikan DataFrame lengkap, dtypes yang benar adalah dtypes asli.
  correct_dtypes = df.dtypes.copy()
  
  result_df = df.mf.apply_chunk(
      custom_set_item,
      output_type="dataframe",
      dtypes=correct_dtypes,
      index=df.index
  )
  
  final_result = result_df.execute().fetch()
  
  print("\nSetelah menggunakan dtypes yang benar, hasil eksekusi adalah sebagai berikut:")
  print(final_result)
  
  session.destroy()

Atur batch_rows untuk mengontrol memori dan konkurensi

• Menghindari error kehabisan memori (OOM) pada satu tugas.

• Meningkatkan tingkat paralelisme.

• Memungkinkan pemanfaatan sumber daya yang lebih baik dengan @with_running_options(memory=...).

Kiat debugging: Cetak hasil antara dan tangkap exception

Karena UDF dijalankan pada worker MaxCompute remote, Anda dapat melihat log print standar di LogView.

Anda dapat mengatur flush=True untuk memastikan log segera ditampilkan guna mempermudah troubleshooting.

def process(chunk):
    try:
        print(f"Memproses chunk dengan bentuk: {chunk.shape}", flush=True)
        print(f"Kolom: {list(chunk.columns)}", flush=True)
        result = chunk.sort_values("B")
        print("Berhasil.", flush=True)
        return result
    except Exception as e:
        print(f"[ERROR] Gagal memproses chunk: {str(e)}", flush=True)
        raise
        

Rekomendasi penyetelan performa

• Ukuran batch: Atur batch_rows berdasarkan volume data dan sumber daya yang tersedia. Hindari mengaturnya ke nilai yang terlalu besar.

• Jumlah kolom output: Hanya kembalikan bidang yang diperlukan.

• Kompleksitas fungsi: Hindari komputasi berat dalam UDF.

• Untuk dependensi eksternal, Anda dapat menggunakan @with_running_options(memory=16) untuk menambah memori.

TypeError: cannot determine dtype

• Penyebab

  Parameter dtypes tidak disediakan.

• Solusi

  Berikan secara eksplisit tipe pd.Series.

Output kosong atau kolom hilang

• Penyebab

  dtypes tidak sesuai.

• Solusi

  Pastikan nama kolom dalam nilai kembali fungsi konsisten.

Eksekusi timeout atau hang

• Penyebab

  Nilai batch_rows terlalu besar.

• Solusi

  Kurangi ukuran batch dan alokasikan lebih banyak sumber daya.