All Products
Search

This Product

    MaxCompute:Kembangkan pekerjaan PyODPS 3

    Document Center

    MaxCompute:Kembangkan pekerjaan PyODPS 3

    Last Updated:Jun 22, 2026

    DataWorks menyediakan node PyODPS 3 untuk menulis dan menjalankan pekerjaan MaxCompute secara berkala dalam Python. Topik ini menjelaskan cara mengonfigurasi dan menjadwalkan pekerjaan Python menggunakan DataWorks.

    Prasyarat

    Node PyODPS 3 telah dibuat. Untuk informasi selengkapnya, lihat Buat dan kelola node MaxCompute.

    Informasi latar belakang

    PyODPS adalah SDK Python untuk MaxCompute yang menyediakan antarmuka pemrograman Python untuk menulis pekerjaan MaxCompute, melakukan kueri terhadap tabel dan view, serta mengelola resource. Untuk informasi selengkapnya, lihat PyODPS. Di DataWorks, Anda dapat menggunakan node PyODPS untuk menjadwalkan dan menjalankan pekerjaan Python serta mengintegrasikannya dengan jenis pekerjaan lainnya.

    Catatan penggunaan

    • Jika kode PyODPS Anda memerlukan paket pihak ketiga, Anda dapat menginstalnya dengan menggunakan serverless resource group dan custom image.

      Catatan

      Jika kode Anda mencakup User-Defined Function (UDF) yang mereferensikan paket pihak ketiga, metode ini tidak didukung. Untuk konfigurasi yang benar, lihat Contoh UDF: Gunakan paket pihak ketiga dalam UDF Python.

    • Untuk melakukan upgrade versi PyODPS, gunakan custom image untuk menjalankan perintah /home/tops/bin/pip3 install pyodps==0.12.1 pada serverless resource group (Anda dapat mengganti 0.12.1 dengan versi PyODPS target), atau gunakan O&M Assistant untuk menjalankan perintah yang sama pada exclusive resource group for scheduling.

    • Jika pekerjaan PyODPS Anda perlu mengakses lingkungan jaringan khusus, seperti sumber data atau layanan di VPC atau pusat data lokal (IDC), gunakan serverless resource group dan bangun koneksi jaringan antara resource group tersebut dengan lingkungan target. Untuk informasi selengkapnya, lihat Solusi konektivitas jaringan.

    • Untuk informasi lebih lanjut mengenai sintaksis PyODPS, lihat dokumentasi PyODPS.

    • Node PyODPS tersedia dalam dua tipe: PyODPS 2 dan PyODPS 3. Keduanya menggunakan versi Python dasar yang berbeda: node PyODPS 2 menggunakan Python 2, sedangkan node PyODPS 3 menggunakan Python 3. Pastikan Anda membuat tipe node yang sesuai dengan versi Python Anda.

    • Jika eksekusi SQL di node PyODPS gagal menghasilkan alur data yang benar sehingga tidak muncul di Peta Data, atasi masalah tersebut dengan mengatur parameter penjadwalan dan waktu proses DataWorks secara manual dalam kode Anda. Untuk mempelajari cara melihat alur data, lihat Melihat alur data. Untuk informasi tentang pengaturan parameter, lihat Mengatur petunjuk parameter waktu proses. Anda dapat memperoleh parameter waktu proses yang diperlukan menggunakan kode contoh berikut:

      import os
...
# get DataWorks sheduler runtime parameters
skynet_hints = {}
for k, v in os.environ.items():
    if k.startswith('SKYNET_'):
        skynet_hints[k] = v
...
# setting hints while submiting a task
o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)
...

    • Ukuran maksimum log output untuk node PyODPS adalah 4 MB. Hindari mencetak hasil data besar ke log. Sebagai gantinya, output hanya informasi alert dan progres yang penting.

    Batasan

    • Saat menjalankan node PyODPS pada exclusive resource group for scheduling, jangan memproses data lokal lebih dari 50 MB. Hal ini disebabkan oleh spesifikasi resource dari exclusive resource group tersebut. Memproses jumlah data lokal yang besar melebihi ambang batas sistem operasi dapat menyebabkan error out-of-memory (OOM), yang ditandai dengan pesan Got Killed. Hindari menulis kode pemrosesan data yang luas secara langsung di node PyODPS. Untuk informasi selengkapnya, lihat Praktik terbaik penggunaan PyODPS yang efisien.

    • Saat Anda menjalankan node PYODPS menggunakan serverless resource group, Anda dapat mengonfigurasi CU untuk node tersebut berdasarkan jumlah data yang perlu diproses.

      Catatan

      Saat menjalankan task di serverless resource group, satu task mendukung konfigurasi maksimum 64CU, tetapi kami menyarankan agar Anda tidak melebihi 16CU untuk mencegah kekurangan resource akibat nilai CU yang berlebihan, yang dapat memengaruhi startup task.

    • Error Got killed menunjukkan bahwa penggunaan memory melebihi batas, sehingga proses dihentikan. Untuk mencegah hal ini, hindari operasi data lokal. Batasan ini tidak berlaku untuk pekerjaan SQL atau DataFrame (kecuali to_pandas) yang diinisiasi melalui PyODPS.

    • Anda dapat menggunakan library Numpy dan Pandas yang telah dipra-instal untuk kode yang tidak melibatkan fungsi kustom. Paket pihak ketiga lain yang berisi kode biner tidak didukung.

    • Karena alasan kompatibilitas, options.tunnel.use_instance_tunnel secara default diatur ke False di DataWorks. Jika Anda perlu mengaktifkan instance tunnel secara global, Anda harus mengatur nilai ini ke True secara manual.

    • Definisi bytecode berbeda antar versi minor Python 3, seperti Python 3.8 dan Python 3.7.

      MaxCompute saat ini menggunakan Python 3.7. Error eksekusi akan terjadi jika Anda menggunakan sintaksis dari versi Python 3 lain, seperti blok finally di Python 3.8. Kami menyarankan agar Anda menggunakan Python 3.7.

    • PyODPS 3 mendukung eksekusi pada serverless resource group. Untuk membeli dan menggunakannya, lihat Gunakan serverless resource groups.

    • Menjalankan beberapa pekerjaan Python secara konkuren dalam satu node PyODPS tidak didukung.

    Edit kode: Contoh dasar

    Setelah Anda membuat node PyODPS, Anda dapat mengedit dan menjalankan kode Anda. Untuk informasi lebih lanjut mengenai sintaksis PyODPS, lihat Ikhtisar operasi dasar.

    • Titik masuk ODPS

      Node PyODPS DataWorks menyediakan variabel global bernama odps atau o sebagai titik masuk ODPS. Anda tidak perlu mendefinisikannya secara manual.

      print(odps.exist_table('PyODPS_iris'))

    • Eksekusi SQL

      Anda dapat mengeksekusi pernyataan SQL di node PyODPS. Untuk informasi selengkapnya, lihat SQL.

      • Secara default, instance tunnel dinonaktifkan di DataWorks. Artinya, instance.open_reader menggunakan antarmuka Result, yang membaca maksimal 10.000 record. Anda dapat menggunakan reader.count untuk mendapatkan jumlah record. Untuk mengiterasi seluruh data, Anda harus menonaktifkan limit. Gunakan pernyataan berikut untuk mengaktifkan instance tunnel secara global dan menonaktifkan limit.

        options.tunnel.use_instance_tunnel = True
options.tunnel.limit_instance_tunnel = False  # Disable the limit to read all data.
with instance.open_reader() as reader:
  # All data can be read through the instance tunnel.

      • Anda juga dapat menambahkan tunnel=True ke open_reader untuk mengaktifkan instance tunnel untuk pemanggilan open_reader saat ini. Demikian pula, Anda dapat menambahkan limit=False untuk menonaktifkan pembatasan limit untuk pemanggilan saat ini.

        # Use the Instance Tunnel interface for the current open_reader operation to read all data.
with instance.open_reader(tunnel=True, limit=False) as reader:

    • Parameter waktu proses

      • Atur parameter waktu proses menggunakan parameter hints, yang merupakan dict. Untuk informasi selengkapnya mengenai hints, lihat Operasi SET.

        o.execute_sql('select * from PyODPS_iris', hints={'odps.sql.mapper.split.size': 16})

      • Jika Anda mengatur sql.settings dalam konfigurasi global, parameter waktu proses ini akan ditambahkan ke setiap eksekusi.

        from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from PyODPS_iris')  # This call includes hints from the global configuration.

    • Hasil eksekusi

      Instans eksekusi SQL dapat langsung menjalankan operasi open_reader dalam dua skenario berikut:

      • Pernyataan SQL mengembalikan data terstruktur.

        with o.execute_sql('select * from dual').open_reader() as reader:
	for record in reader:  # Process each record.

      • Saat mengeksekusi pernyataan seperti desc, Anda dapat mengambil hasil eksekusi SQL mentah menggunakan properti reader.raw.

        with o.execute_sql('desc dual').open_reader() as reader:
	print(reader.raw)
        Catatan

        Jika Anda menggunakan parameter penjadwalan kustom, Anda harus menuliskan waktu secara eksplisit saat memicu langsung eksekusi node PyODPS 3 dari halaman. Node PyODPS tidak dapat langsung menggantikan nilai ini.

    • DataFrame

      Anda juga dapat memproses data menggunakan DataFrame (tidak direkomendasikan).

      • Eksekusi

        Di lingkungan DataWorks, operasi DataFrame harus dipicu secara eksplisit dengan memanggil metode yang langsung dieksekusi.

        from odps.df import DataFrame
iris = DataFrame(o.get_table('pyodps_iris'))
for record in iris[iris.sepal_width < 3].execute():  # Call an immediately executed method to process each record.

        Jika Anda perlu memicu eksekusi segera saat mencetak, Anda harus mengaktifkan options.interactive.

        from odps import options
from odps.df import DataFrame
options.interactive = True  # Enable the switch at the beginning.
iris = DataFrame(o.get_table('pyodps_iris'))
print(iris.sepal_width.sum())  # An immediate execution is triggered when printing.

      • Cetak informasi detail

        Atur opsi options.verbose. Opsi ini secara default diaktifkan di DataWorks dan mencetak informasi detail, seperti URL Logview, selama eksekusi.

    Contoh

    Contoh berikut menunjukkan cara menggunakan node PyODPS:

    1. Siapkan set data dan buat tabel sampel pyodps_iris. Untuk detailnya, lihat Memproses data dengan menggunakan DataFrame.

    2. Buat DataFrame. Untuk detailnya, lihat Membuat DataFrame dari tabel MaxCompute.

    3. Masukkan dan jalankan kode berikut di node PyODPS.

      from odps.df import DataFrame
# Create a DataFrame from an ODPS table.
iris = DataFrame(o.get_table('pyodps_iris'))
print(iris.sepallength.head(5))

      Hasil berikut dikembalikan:

         sepallength
0          4.5
1          5.5
2          4.9
3          5.0
4          6.0

    Edit kode: Contoh lanjutan

    Jika node perlu dijalankan secara berkala, Anda harus menentukan properti penjadwalannya. Untuk informasi selengkapnya, lihat Konfigurasi properti penjadwalan untuk node.

    Parameter penjadwalan

    Di panel kanan editor node, klik Scheduling Settings. Di bagian Parameter, konfigurasi parameter kustom. Cara mendefinisikan variabel di node PyODPS berbeda dengan cara mendefinisikannya di node SQL. Untuk informasi selengkapnya, lihat Konfigurasi parameter penjadwalan.

    Berbeda dengan node SQL di DataWorks, node PyODPS tidak mengganti string seperti ${param_name} dalam kode. Sebagai gantinya, sebuah dict bernama args ditambahkan ke variabel global sebelum kode dieksekusi. Anda dapat mengambil parameter penjadwalan dari dict ini. Misalnya, jika Anda mengatur ds=${yyyymmdd} di Parameter, Anda dapat menggunakan metode berikut untuk mengambil parameter ini dalam kode Anda.

    print('ds=' + args['ds'])
ds=20161116
    Catatan

    Jika Anda perlu mendapatkan partisi bernama ds, Anda dapat menggunakan metode berikut.

    o.get_table('table_name').get_partition('ds=' + args['ds'])

    Untuk informasi selengkapnya mengenai pengembangan pekerjaan PyODPS untuk kasus penggunaan lain, lihat topik-topik berikut:

    Langkah selanjutnya

    FAQ

    T: Saya menggunakan node PyODPS3 untuk mengumpulkan data dari API pihak ketiga, seperti Lark, dan mengimpornya ke DataWorks. Kode berjalan tanpa masalah di lingkungan pengembangan lokal saya, tetapi melaporkan error timeout respons saat dikirim ke lingkungan produksi dan dieksekusi di Operation Center. Mengapa?

    J: Di Management Center > Security Settings > Workspace Anda. Di daftar putih sandbox, tambahkan nama domain API pihak ketiga untuk memberikan akses ke pekerjaan PyODPS 3. Contohnya:

    Tambahkan nama domain API Lark open.feishu.cn dan atur port ke 443.