A MaxCompute PyFG (Python Feature Generation) job generates complex features in offline batches. It supports complex ODPS 2.0 data types, including list, map, float, and int. The job uses a configuration file and command-line parameters to determine whether to bin the generated features.
Method 1: Use a general-purpose resource group image
In the DataWorks console, go to Scheduling Configuration > Resource Properties. Select a general-purpose resource group and choose the latest dataworks_pairec_task_pod image.
Note: The release of the dataworks_pairec_task_pod image may lag behind pyfg updates, so the image might not contain the latest pyfg package. To check the required version, see the script generated by Recommended Solution Customization - Feature Configuration. To use the latest version of pyfg, use Method 3 to create a custom image for your resource group.
Method 2: Install dependencies (for older versions of DataWorks)
Log in to the DataWorks console, create an exclusive resource group for scheduling, and use O&M Assistant to install the pyfg package.
To install the pyfg package in your DataWorks exclusive resource group, navigate to DataWorks->Management Center->Resource Group List->O&M Assistant and run the following command:
/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.whlFAQ
Method 3: Customize a resource group image (for new versions of DataWorks)
For instructions, see Custom images.
Upload resource files
Upload the FG configuration file (in JSON format) to your MaxCompute project.
Some feature operators require additional resource files. You must manually upload these files to your MaxCompute project.
Feature operator | Description | Resource file parameter |
Text normalization | Stop word file | |
Text tokenization feature | Vocabulary configuration file | |
Text relevance feature | Term frequency configuration file | |
custom operator | Operator configuration file |
Create the output table
In DataWorks, create a PyOdps3 node and run the following script. The script reads fg.json and creates the output table and the resources required for subsequent runs.
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)Before you run the script, configure the following parameters in Scheduling Configuration: input_table, output_table, fg_json_file, and partition_value.
Although fg_task.run(o) also creates the output table automatically if it does not exist, we recommend that you call this method to create the output table in advance. This helps prevent conflicts and task failures when you backfill data concurrently.
Run the FG offline task
In DataWorks, create a PyOdps3 node and run the following script. The script runs the feature generation task and automatically creates the output table if it does not exist.
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)
Before you run the script, configure the following parameters in Scheduling Configuration: input_table, output_table, fg_json_file, and partition_value.
If you have PyODPS installed, you can also install pyfg and submit tasks locally.
Parameters
Parameter | Default | Description |
input_table | None | The input table. |
output_table | None | The output table. Created automatically if it does not exist. |
fg_json_file | None | The FG configuration file, in JSON format. |
partition_value | None | The input table partition to process. Results are written to the corresponding partition in the output table. |
schema | None | The MaxCompute schema. For more information, see schema operations. |
batch_size | 128 | The number of records to process in each batch. |
memory | 1024 | The amount of memory to allocate to the task node, in MiB. |
force_delete_output_table | False | If set to True, the system deletes the output table before it runs the task. |
force_update_resource | False | If set to True, the system updates resources before it runs the task. To prevent concurrency conflicts, avoid leaving this parameter set to True. |
output_merged_str | False | If set to True, strings are automatically merged to output a large string feature in RTP format. |
debug | False | If set to True, the task runs in debug mode and prints the content of all updated resources. |
sql_setting | None | MaxCompute SQL parameters. Use the |
fg_setting | None | FG parameters. Use the |
You can override the default parameter values by passing them to the FgTask constructor, as shown in the examples.
How it works
The pyfg package runs on a gateway machine within an exclusive resource group to submit SQL tasks to MaxCompute. Alternatively, you can install both pyfg and PyODPS on a local machine to submit tasks.
custom UDFs in an SQL task require resources such as the FG shared library, configuration files (for example, fg.json, dictionaries, and custom operator libraries), and the UDF code files (.py). All these resources must be uploaded to the MaxCompute cluster and stored in its distributed file system. When a task runs, each worker downloads the required resources from the distributed file system and loads them into memory.
Some resources, such as the FG shared library and UDF code files, are shared across multiple tasks. When force_update_resource=True, the system deletes the original resources before uploading the new ones. This process creates a time window that can disrupt other running tasks.