MaxCompute：Mengacu pada paket pihak ketiga dalam node PyODPS

Sebelum merujuk ke paket pihak ketiga, pastikan bahwa paket tersebut telah diunggah ke MaxCompute sebagai sumber daya arsip. Anda dapat mengunggah paket pihak ketiga menggunakan salah satu metode berikut:

• Gunakan kode untuk mengunggah paket pihak ketiga. Dalam contoh kode berikut, ganti packages.tar.gz dengan jalur dan nama paket yang ingin Anda unggah.

  import os
  from odps import ODPS
  
  # Setel variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET ke ID AccessKey dan Rahasia AccessKey akun Alibaba Cloud Anda.
  # Kami menyarankan agar Anda tidak langsung menggunakan ID AccessKey atau Rahasia AccessKey Anda.
  o = ODPS(
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
      project='<proyek-default-anda>',
      endpoint='<titik-akhir-anda>',
  )
  o.create_resource("test_packed.tar.gz", "archive", fileobj=open("packages.tar.gz", "rb"))

• Gunakan DataWorks untuk mengunggah paket pihak ketiga. Untuk informasi lebih lanjut, lihat Langkah 1: Buat Sumber Daya atau Unggah Sumber Daya yang Ada.

Sebelum merujuk ke paket pihak ketiga dalam fungsi terdefinisi pengguna (UDF) Python, modifikasi UDF Python tersebut sesuai prosedur berikut:

1. Tambahkan referensi ke paket pihak ketiga dalam metode __init__ dari kelas UDF.

2. Referensikan paket pihak ketiga menggunakan kode UDF seperti fungsi evaluate atau metode process.

Dalam contoh ini, sebuah paket pihak ketiga dirujuk menggunakan Python UDF bernama psi dalam SciPy.

1. Jalankan perintah berikut untuk membungkus SciPy.

   pyodps-pack -o scipy-bundle.tar.gz scipy

2. Tulis kode berikut dan simpan sebagai file bernama test_psi_udf.py.

   import sys
   from odps.udf import annotate
   
   @annotate("double->double")
   class MyPsi(object):
       def __init__(self):
           # Tambahkan jalur ke jalur referensi.
           sys.path.insert(0, "work/scipy-bundle.tar.gz/packages")
   
       def evaluate(self, arg0):
           # Letakkan pernyataan IMPORT di dalam tubuh fungsi evaluate.
           from scipy.special import psi
   
           return float(psi(arg0))

   Deskripsi kode: Fungsi __init__ menambahkan work/scipy-bundle.tar.gz/packages ke sys.path. Ini karena MaxCompute mengekstraksi semua sumber daya arsip yang dirujuk oleh UDF ke folder di direktori work. Nama folder sama dengan nama sumber daya. Direktori packages adalah subdirektori dari paket yang dihasilkan dengan menggunakan pyodps-pack. Pernyataan IMPORT SciPy ditempatkan di dalam tubuh fungsi evaluate. Ini karena paket pihak ketiga hanya tersedia selama waktu proses. Saat UDF diproses di server MaxCompute, lingkungan pemrosesan tidak berisi paket pihak ketiga. Jika Anda meletakkan pernyataan IMPORT di luar tubuh fungsi untuk mengimpor paket pihak ketiga, kesalahan akan dilaporkan.

3. Unggah test_psi_udf.py sebagai sumber daya Python MaxCompute dan unggah scipy-bundle.tar.gz sebagai sumber daya arsip.

4. Buat UDF bernama test_psi_udf, referensikan dua file sumber daya yang diunggah, dan tentukan nama kelas sebagai test_psi_udf.MyPsi.

   Anda dapat melakukan Langkah 3 dan Langkah 4 di node PyODPS atau di klien MaxCompute.

   • Di node PyODPS:

     import os
     from odps import ODPS
     
     # Setel variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET ke ID AccessKey dan Rahasia AccessKey akun Alibaba Cloud Anda.
     # Kami menyarankan agar Anda tidak langsung menggunakan ID AccessKey atau Rahasia AccessKey Anda.
     
     o = ODPS(
         os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
         os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
         project='<proyek-default-anda>',
         endpoint='<titik-akhir-anda>',
     )
     
     bundle_res = o.create_resource(
         "scipy-bundle.tar.gz", "archive", fileobj=open("scipy-bundle.tar.gz", "rb")
     )
     udf_res = o.create_resource(
         "test_psi_udf.py", "py", fileobj=open("test_psi_udf.py", "rb")
     )
     o.create_function(
         "test_psi_udf", class_type="test_psi_udf.MyPsi", resources=[bundle_res, udf_res]
     )

   • Di klien MaxCompute:

     add archive scipy-bundle.tar.gz;
     add py test_psi_udf.py;
     create function test_psi_udf as test_psi_udf.MyPsi using test_psi_udf.py,scipy-bundle.tar.gz;

5. Setelah menyelesaikan operasi sebelumnya, gunakan UDF untuk mengeksekusi pernyataan SQL.

   set odps.pypy.enabled=false;
   set odps.isolation.session.enable=true;
   select test_psi_udf(sepal_length) from iris;

Anda dapat merujuk pustaka pihak ketiga dalam DataFrame PyODPS dengan menentukan parameter libraries dalam metode execute atau persist. Bagian ini menjelaskan cara merujuk paket pihak ketiga dalam DataFrame PyODPS jika Anda menggunakan metode map. Prosedurnya serupa jika Anda menggunakan metode apply atau map_reduce.

1. Jalankan perintah berikut untuk membungkus SciPy:

   pyodps-pack -o scipy-bundle.tar.gz scipy

2. Tabel bernama test_float_col tersedia. Tabel ini hanya berisi satu kolom tipe FLOAT.

      col1
   0  3.75
   1  2.51

   Jalankan kode berikut untuk menghitung nilai dari psi(col1):

   import os
   from odps import ODPS, options
   
   def my_psi(v):
       from scipy.special import psi
   
       return float(psi(v))
   
   # Jika fitur isolasi diaktifkan untuk proyek Anda, Anda tidak diharuskan mengonfigurasi opsi berikut:
   options.sql.settings = {"odps.isolation.session.enable": True}
   
   # Setel variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET ke ID AccessKey dan Rahasia AccessKey akun Alibaba Cloud Anda.
   # Kami menyarankan agar Anda tidak langsung menggunakan ID AccessKey atau Rahasia AccessKey Anda.
   
   o = ODPS(
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
       os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
       project='<proyek-default-anda>',
       endpoint='<titik-akhir-anda>',
   )
   df = o.get_table("test_float_col").to_df()
   # Jalankan kode berikut dan peroleh hasilnya.
   df.col1.map(my_psi).execute(libraries=["scipy-bundle.tar.gz"])
   # Simpan hasilnya ke tabel lain.
   df.col1.map(my_psi).persist("result_table", libraries=["scipy-bundle.tar.gz"])

3. Opsional. Jika Anda ingin menggunakan paket pihak ketiga yang sama selama waktu proses, Anda dapat mengonfigurasi parameter global.

   from odps import options
   options.df.libraries = ["scipy-bundle.tar.gz"]

Setelah melakukan operasi sebelumnya, rujuk paket pihak ketiga dalam DataFrame PyODPS.

Dalam beberapa proyek yang sudah ada, Anda mungkin mengunggah semua dependensi WHL secara manual dan merujuknya dalam kode, atau menggunakan MaxCompute versi awal yang tidak mendukung paket biner. Bagian ini menjelaskan cara mengunggah dan merujuk paket pihak ketiga secara manual dalam proyek yang sudah ada ini. Dalam contoh ini, python_dateutil digunakan dalam metode map.

1. Jalankan perintah pip download di Linux Bash untuk mengunduh paket pihak ketiga dan dependensinya ke direktori. Dua paket diunduh: six-1.10.0-py2.py3-none-any.whl dan python_dateutil-2.5.3-py2.py3-none-any.whl.

   pip download python-dateutil -d /to/path/

   Catatan

   Perhatikan bahwa paket yang diunduh harus mendukung sistem operasi Linux. Kami menyarankan agar Anda menjalankan perintah ini di sistem operasi Linux.

2. Unggah paket yang diunduh ke MaxCompute.

   • Metode 1: Gunakan kode.

     # Anda harus memastikan bahwa ekstensi nama file valid.
     odps.create_resource('six.whl', 'file', file_obj=open('six-1.10.0-py2.py3-none-any.whl', 'rb'))
     odps.create_resource('python_dateutil.whl', 'file', file_obj=open('python_dateutil-2.5.3-py2.py3-none-any.whl', 'rb'))

   • Metode 2: Gunakan DataWorks.

     Anda dapat mengunggah dan mengirimkan sumber daya tujuan dengan mengikuti petunjuk dalam Langkah 1: Buat Sumber Daya atau Unggah Sumber Daya yang Ada.

3. Rujuk paket pihak ketiga.

   Dalam contoh ini, objek DataFrame hanya berisi satu bidang tipe STRING. Bidang tersebut berisi konten berikut.

                  datestr
   0  2016-08-26 14:03:29
   1  2015-08-26 14:03:29

   • Gunakan pustaka pihak ketiga berikut dalam konfigurasi global:

     from odps import options
     
     def get_year(t):
         from dateutil.parser import parse
         return parse(t).strftime('%Y')
     
     options.df.libraries = ['six.whl', 'python_dateutil.whl']
     df.datestr.map(get_year).execute()

        datestr
     0     2016
     1     2015

   • Gunakan parameter libraries dari metode yang dipanggil langsung untuk menentukan pustaka:

     def get_year(t):
         from dateutil.parser import parse
         return parse(t).strftime('%Y')
     
     df.datestr.map(get_year).execute(libraries=['six.whl', 'python_dateutil.whl'])

        datestr
     0     2016
     1     2015

Secara default, PyODPS mendukung pustaka Python yang hanya berisi kode Python dan tidak melibatkan operasi file. Di versi MaxCompute yang lebih baru, PyODPS juga mendukung pustaka Python yang berisi kode biner atau melibatkan operasi file. Nama pustaka harus memiliki akhiran. Tabel berikut menjelaskan akhiran yang didukung oleh pustaka Python yang berbeda.

Semua paket WHL harus diunggah ke MaxCompute sebagai sumber daya arsip. Sebelum mengunggah paket, Anda harus mengonversi paket menjadi file ZIP dengan mengubah ekstensi nama file. Anda juga perlu menyetel parameter odps.isolation.session.enable ke True untuk pekerjaan atau proyek Anda. Contoh berikut menunjukkan cara mengunggah dan menggunakan fungsi khusus dalam SciPy:

# Paket yang berisi kode biner harus diunggah sebagai sumber daya arsip. Anda harus mengonversi paket WHL menjadi file ZIP sebelum Anda mengunggah paket.
odps.create_resource('scipy.zip', 'archive', file_obj=open('scipy-0.19.0-cp27-cp27m-manylinux1_x86_64.whl', 'rb'))

# Jika fitur isolasi diaktifkan untuk proyek Anda, Anda tidak diharuskan mengonfigurasi opsi berikut:
options.sql.settings = { 'odps.isolation.session.enable': True }

def my_psi(value):
    # Kami menyarankan agar Anda meletakkan pernyataan IMPORT di dalam fungsi untuk mengimpor pustaka pihak ketiga. Ini mencegah kesalahan waktu proses yang disebabkan oleh perbedaan struktural paket biner di sistem operasi yang berbeda.
    from scipy.special import psi
    return float(psi(value))

df.float_col.map(my_psi).execute(libraries=['scipy.zip'])

Anda dapat membungkus paket biner yang hanya berisi kode sumber ke dalam file WHL dengan menjalankan perintah shell berikut di Linux dan kemudian mengunggahnya. File WHL yang dihasilkan di macOS atau Windows tidak dapat digunakan di MaxCompute.

python setup.py bdist_wheel