Konfigurasi dan jalankan task batch FG di MaxCompute melalui PyFG - Artificial Intelligence Recommendation - Alibaba Cloud -

Pekerjaan MaxCompute PyFG (Python Feature Generation) menghasilkan fitur kompleks secara batch offline dan mendukung tipe data ODPS 2.0 yang kompleks, seperti list, map, float, dan int. Pekerjaan ini menggunakan file konfigurasi serta parameter command-line untuk menentukan apakah akan melakukan binning terhadap fitur yang dihasilkan.

Metode 1: Gunakan gambar kelompok sumber daya serbaguna

Di Konsol DataWorks, buka Scheduling Configuration > Resource Properties. Pilih kelompok sumber daya serbaguna dan gunakan gambar dataworks_pairec_task_pod versi terbaru.

Catatan: Rilis gambar dataworks_pairec_task_pod mungkin tertinggal dari pembaruan pyfg, sehingga gambar tersebut belum tentu berisi paket pyfg terbaru. Untuk memeriksa versi yang diperlukan, lihat skrip yang dihasilkan oleh Penyesuaian Solusi yang Direkomendasikan - Konfigurasi Fitur. Untuk menggunakan versi terbaru pyfg, gunakan Metode 3 untuk membuat custom image pada kelompok sumber daya Anda.

Metode 2: Instal dependensi (untuk versi lama `DataWorks`)

Masuk ke Konsol DataWorks, buat grup sumber daya eksklusif untuk penjadwalan, lalu gunakan O&M Assistant untuk menginstal paket pyfg.

Untuk menginstal paket pyfg di kelompok sumber daya eksklusif DataWorks Anda, buka DataWorks->Management Center->Resource Group List-><a href="https://dataworks.console.alibabacloud.com/resource/runcommand" id="963b1e2ebd2ts">O&M Assistant</a>, lalu jalankan perintah berikut:

/home/tops/bin/pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple --upgrade --force-reinstall http://tzrec.oss-cn-beijing.aliyuncs.com/third_party/pyfg105-1.0.5-cp37-cp37m-linux_x86_64.whl

FAQ

Jika Anda menemui error berikut:

Executing user script with PyODPS 0.10.8 (wrapper version: 0.11.6)
/home/tops/lib/python3.7/site-packages/odps/types.py:1381: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  float_builtins += (np.float,)
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:35: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_float_types = list(map(np.dtype, [np.float, np.float32, np.float64]))
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:40: DeprecationWarning: `np.bool` is a deprecated alias for the builtin `bool`. To silence this warning, use `bool` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.bool_` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_to_df_types[np.dtype(np.bool)] = types.boolean
/home/tops/lib/python3.7/site-packages/odps/df/backends/pd/types.py:41: DeprecationWarning: `np.str` is a deprecated alias for the builtin `str`. To silence this warning, use `str` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.str_` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  _np_to_df_types[np.dtype(np.str)] = types.string
/home/tops/lib/python3.7/site-packages/odps/ipython/magics.py:35: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  np_float_types = map(np.dtype, [np.float, np.float16, np.float32, np.float64])
package: pyfg105
package path: /home/tops/lib/python3.7/site-packages/pyfg105
custom_lib_path: /home/tops/lib/python3.7/site-packages/pyfg105/lib
Initializing pyfg...
WARNING: Logging before InitGoogleLogging() is written to STDERR
I20260525 20:54:34.598881    21 str_utils.cc:252] GlobalConfig ctor: 0x7fc8a7ffa040
I20260525 20:54:34.598944    21 str_utils.cc:54] AVX supported
I20260525 20:54:34.598951    21 str_utils.cc:56] FMA supported
I20260525 20:54:34.598958    21 str_utils.cc:75] AVX-512F not supported
I20260525 20:54:34.598961    21 str_utils.cc:255] support avx512: false
I20260525 20:54:34.598968    21 str_utils.cc:264] will NOT use avx512
I20260525 20:54:34.598982    21 base_feature.cc:705] reset global fg config
pyfg version: 1.0.5 loaded
2026-05-25 20:54:34,606 WARNING:odps.pyodpswrapper:Code with error
===================
from pyfg105 import run_on_odps
fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    force_delete_output_table=True,
    force_update_resource=True)
fg_task.create_output_table(o)
===================
Traceback (most recent call last):
  File "/home/tops/lib/python3.7/site-packages/pyfg105/run_on_odps.py", line 12, in <module>
    from .create_output_table import TableCreator
  File "/home/tops/lib/python3.7/site-packages/pyfg105/create_output_table.py", line 2, in <module>
    from odps.models import TableSchema, Column
ImportError: cannot import name 'TableSchema' from 'odps.models' (/home/tops/lib/python3.7/site-packages/odps/models/__init__.py)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "<pyodps_user_code>", line 1, in <module>
    from pyfg105 import run_on_odps
  File "/home/tops/lib/python3.7/site-packages/pyfg105/run_on_odps.py", line 15, in <module>
    from create_output_table import TableCreator
  File "/home/tops/lib/python3.7/site-packages/pyfg105/create_output_table.py", line 2, in <module>
    from odps.models import TableSchema, Column
ImportError: cannot import name 'TableSchema' from 'odps.models' (/home/tops/lib/python3.7/site-packages/odps/models/__init__.py)

Error ini terjadi karena versi pyodps pada kelompok sumber daya eksklusif saat ini sudah usang. Anda perlu memperbaruinya ke versi terbaru dan juga memperbarui paket pyodpswrapper yang sesuai. Namun, tindakan ini dapat mengubah versi paket lain di lingkungan python, yang berpotensi menyebabkan tugas python Anda gagal. Pendekatan paling aman adalah menggunakan custom image untuk kelompok sumber daya Anda (Metode 3).

Metode 3: Sesuaikan gambar kelompok sumber daya (untuk versi baru `DataWorks`)

Untuk petunjuknya, lihat Custom images.

Unggah file sumber daya

Unggah file konfigurasi FG (dalam format JSON) ke proyek MaxCompute Anda.

Beberapa operator fitur memerlukan file sumber daya tambahan. Anda harus mengunggah file-file tersebut secara manual ke proyek MaxCompute Anda.

Feature operator	Description	Resource file parameter
text_normalizer	Text normalization	File stop word `stop_char_file`
tokenize_feature	Text tokenization feature	File konfigurasi vocabulary `vocab_file`
bm25_feature	Text relevance feature	File konfigurasi frekuensi term `term_doc_freq_file`
custom_feature	custom operator	File konfigurasi operator `operator_lib_file`

Buat tabel output

Di DataWorks, buat node PyOdps3 dan jalankan skrip berikut untuk membaca fg.json, lalu membuat tabel output serta sumber daya yang diperlukan untuk eksekusi selanjutnya.

from pyfg105 import run_on_odps

fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    force_delete_output_table=True,
    force_update_resource=True)
fg_task.create_output_table(o)

Sebelum menjalankan skrip, konfigurasikan parameter berikut di Scheduling Configuration: input_table, output_table, fg_json_file, dan partition_value.

Meskipun fg_task.run(o) secara otomatis membuat tabel output jika belum ada, kami merekomendasikan agar Anda memanggil metode ini terlebih dahulu untuk membuat tabel output. Langkah ini membantu mencegah konflik dan kegagalan tugas saat melakukan backfill data secara konkuren.

Jalankan tugas offline FG

Di DataWorks, buat node PyOdps3 dan jalankan skrip berikut. Skrip ini menjalankan tugas pembuatan fitur dan secara otomatis membuat tabel output jika belum ada.

from pyfg105 import run_on_odps

fg_task = run_on_odps.FgTask(
    args['input_table'], 
    args['output_table'], 
    args['fg_json_file'], 
    args['partition_value'],
    batch_size=128,
    force_delete_output_table=False,
    force_update_resource=False)
fg_task.add_sql_setting('odps.stage.mapper.split.size', 256)
fg_task.run(o)

Sebelum menjalankan skrip, konfigurasikan parameter berikut di Scheduling Configuration: input_table, output_table, fg_json_file, dan partition_value.

Jika Anda telah menginstal PyODPS, Anda juga dapat menginstal pyfg dan mengirimkan tugas secara lokal.

Parameter

Parameter	Default	Description
input_table	None	Tabel input.
output_table	None	Tabel output. Dibuat secara otomatis jika belum ada.
fg_json_file	None	File konfigurasi FG, dalam format JSON.
partition_value	None	Partisi tabel input yang akan diproses. Hasil ditulis ke partisi yang sesuai di tabel output.
schema	None	Skema MaxCompute. Untuk informasi lebih lanjut, lihat operasi skema.
batch_size	128	Jumlah record yang diproses dalam setiap batch.
memory	1024	Jumlah memori yang dialokasikan ke node tugas, dalam satuan MiB.
force_delete_output_table	False	Jika diatur ke True, sistem akan menghapus tabel output sebelum menjalankan tugas.
force_update_resource	False	Jika diatur ke True, sistem akan memperbarui sumber daya sebelum menjalankan tugas. Untuk mencegah konflik konkurensi, hindari mengaktifkan parameter ini.
output_merged_str	False	Jika diatur ke True, string akan secara otomatis digabung untuk menghasilkan fitur string besar dalam format RTP.
debug	False	Jika diatur ke True, tugas berjalan dalam mode debug dan mencetak konten semua sumber daya yang diperbarui.
sql_setting	None	Parameter SQL MaxCompute. Gunakan metode `fg_task.add_sql_setting` untuk mengonfigurasinya. Untuk informasi lebih lanjut, lihat Flag parameters. Anda dapat menambahkan beberapa flag.
fg_setting	None	Parameter FG. Gunakan metode `fg_task.add_fg_setting` untuk mengonfigurasinya. Untuk informasi lebih lanjut, lihat [Global configuration](summary.md#id9). Anda dapat menambahkan beberapa item konfigurasi. Tersedia sejak v0.4.0.

Anda dapat mengganti nilai parameter default dengan meneruskannya ke konstruktor FgTask, seperti yang ditunjukkan dalam contoh.

Cara kerja

Paket pyfg berjalan pada mesin gateway dalam kelompok sumber daya eksklusif untuk mengirimkan tugas SQL ke MaxCompute. Alternatifnya, Anda dapat menginstal pyfg dan PyODPS pada mesin lokal untuk mengirimkan tugas.

UDF kustom dalam tugas SQL memerlukan sumber daya seperti library bersama FG, file konfigurasi (misalnya, fg.json, kamus, dan library operator kustom), serta file kode UDF (.py). Semua sumber daya ini harus diunggah ke kluster MaxCompute dan disimpan dalam sistem file terdistribusinya. Saat tugas berjalan, setiap worker mengunduh sumber daya yang diperlukan dari sistem file terdistribusi dan memuatnya ke memori.

Beberapa sumber daya, seperti library bersama FG dan file kode UDF, digunakan bersama oleh beberapa tugas. Ketika force_update_resource=True, sistem menghapus sumber daya asli sebelum mengunggah yang baru. Proses ini menciptakan jendela waktu yang dapat mengganggu tugas lain yang sedang berjalan.