在MaxCompute任務中使用FG離線產生特徵-智能推薦 AIRec-阿里雲 -

MaxCompute PyFg Job可以離線批量產生複雜特徵，支援ODPS2.0的複雜類型（list、map、float、int等），根據設定檔和命令列參數決定是否對產生的特徵做分箱操作。

方式一：使用通用資源群組鏡像

在DataWorks的調度配置-資源屬性區塊選擇一個通用的資源群組，選擇最新的dataworks_pairec_task_pod鏡像。

注意：dataworks_pairec_task_pod鏡像的發布相對於pyfg的版本更新可能比較滯後，最新的鏡像內包含的pyfg包可能不是最新版本（具體版本請參考推薦方案定製-特徵配置產生的指令碼）。可按照下面描述的方式三自訂資源群組鏡像，使用最新的pyfg。

方式二：安裝依賴包（適用於老版`DataWorks`）

登入DataWorks控制台，建立獨享調度資源群組，使用營運助手安裝pyfg的包。

在DataWorks獨享資源群組中安裝pyfg包，頁面路徑為：DataWorks->管理中心->資源群組列表->營運助手，樣本如下：

/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

FQA （如遇執行報錯請看這裡，點擊展開）

如果遇到如下錯誤：

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)

是因為當前獨享資源群組上pyodps的版本太老了，需要更新到最新版本，同時需要同步更新配套的pyodpswrapper包。但是，這樣做可能造成python環境某些包版本變化，進而導致python任務失敗，最穩妥的做法是直接用新版資源群組的自訂鏡像（方式三）。

方式三：自訂資源群組鏡像（適用於新版`DataWorks`）

參考文檔：自訂鏡像。

上傳資源檔

上傳FG的設定檔（JSON格式）到MaxCompute的專案空間中。

某些FG運算元需要額外的資源檔，使用者需要手動上傳到MaxCompute的專案空間中。

特徵運算元	描述	資源檔配置項
text_normalizer	文本歸一化	停用詞檔案`stop_char_file`
tokenize_feature	文本分詞特徵	詞彙表設定檔`vocab_file`
bm25_feature	文本相關性特徵	詞頻設定檔 `term_doc_freq_file`
custom_feature	自訂運算元	運算元設定檔`operator_lib_file`

建立輸出表

可以在DataWorks裡建立一個PyOdps3節點，執行下面的指令碼。程式會根據fg.json的內容建立輸出表，同時建立後續執行任務時需要使用的各種資源。

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)

執行之前需要在調度配置裡配置好參數：input_table、output_table、fg_json_file、partition_value。

雖然fg_task.run(o)本身也會在沒有輸出表時自動建立輸出表，但還是建議先調用該介面預先建立好輸出表，防止在並發補資料時引起衝突導致任務失敗。

執行 FG 離線任務

在DataWorks裡建立一個PyOdps3節點，執行下面的指令碼，程式會根據fg.json的內容自動建立輸出表。

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)

執行之前需要在調度配置裡配置好參數：input_table、output_table、fg_json_file、partition_value。

如果本地安裝了PyODPS，pyfg也可以安裝在本地，在本地提交任務。

參數說明

參數	預設值	說明
input_table	無	輸入表。
output_table	無	輸出表，會自動建立。
fg_json_file	無	FG設定檔，JSON格式。
partition_value	無	指定輸入表的partition分區作為FG的輸入，結果儲存在輸出表的partition分區中。
schema	無	指定[MaxCompute的Schema](Schema操作)
batch_size	128	批量處理的記錄數。
memory	1024	任務節點使用的記憶體大小，單位：M
force_delete_output_table	False	是否刪除輸出表，設定為True時會先自動刪除輸出表，再運行任務。
force_update_resource	False	是否更新資源，設定為True時會先自動更新資源，再運行任務；不要一直設定為True，會有並發衝突的問題。
output_merged_str	False	是否合并字串，設定為 True 時會自動合并字串，輸出RTP格式的大String特徵
debug	False	是否是調試模型，設定為True時會列印所有更新的資源的內容。
sql_setting	無	`fg_task.add_sql_setting`方法的key、value參數。指定MaxCompute SQL參數，詳細見Flag參數列表，可以add多個flag。
fg_setting	無	`fg_task.add_fg_setting`方法的key、value參數。指定 FG 參數，詳細見[全域配置](summary.md#id9)，可以 add 多個配置項；從 v0.4.0 開始生效

直接按照樣本修改代碼中的參數預設值即可。

補充說明

pyfg的包是裝在專有資源群組的一台機器上（gateway）；這台機器可以提交SQL任務到MaxCompute。pyfg的包也可以裝在自己的任何一台機器上，這台機器通過pyodps這個工具就可以提交任務到MaxCompute叢集。

SQL任務裡的自訂UDF需要一些資源，包括FG的共用庫，各種設定檔（如fg.json、詞典、自訂OP的lib等），以及UDF自己的代碼檔案（.py），這些資源都需要上傳到MaxCompute叢集，儲存在MaxCompute的Distributed File System中。執行任務時，每個worker都會從MaxCompute叢集的Distributed File System中下載這些資源，並載入到記憶體中。

有部分資源是多個任務共用的，比如FG的共用庫，UDF的代碼檔案等。force_update_resource=True時，會先刪除原來的資源，再上傳新的資源。這裡有個時間間隔，在這個時間間隔內，會影響到其他正在執行的任務。