Um job do MaxCompute PyFG (Python Feature Generation) gera features complexas em lotes offline. Ele oferece suporte a tipos de dados complexos do ODPS 2.0, incluindo list, map, float e int. O job utiliza um arquivo de configuração e parâmetros de linha de comando para determinar se as features geradas devem ser agrupadas em bins.
Método 1: Usar uma imagem de grupo de recursos de uso geral
No console do DataWorks, acesse Scheduling Configuration > Resource Properties. Selecione um grupo de recursos de uso geral e escolha a versão mais recente da imagem dataworks_pairec_task_pod.
Nota: A liberação da imagem dataworks_pairec_task_pod pode ocorrer com atraso em relação às atualizações do pyfg. Portanto, a imagem pode não conter o pacote pyfg mais recente. Para verificar a versão necessária, consulte o script gerado por Personalização da solução recomendada - Configuração de features. Para utilizar a versão mais recente do pyfg, siga o Método 3 e crie uma imagem personalizada para o seu grupo de recursos.
Método 2: Instalar dependências (para versões anteriores do DataWorks)
Faça login no console do DataWorks, crie um grupo de recursos exclusivo para agendamento e utilize o O&M Assistant para instalar o pacote pyfg.
Para instalar o pacote pyfg no seu grupo de recursos exclusivo do DataWorks, acesse DataWorks->Management Center->Resource Group List->O&M Assistant e execute o seguinte comando:
/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.whlPerguntas frequentes
Método 3: Personalizar uma imagem de grupo de recursos (para novas versões do DataWorks)
Para obter instruções, consulte Imagens personalizadas.
Carregar arquivos de recursos
Carregue o arquivo de configuração do FG (no formato JSON) no seu projeto do MaxCompute.
Alguns operadores de features exigem arquivos de recursos adicionais. Carregue manualmente esses arquivos no seu projeto do MaxCompute.
|
Operador de feature |
Descrição |
Parâmetro do arquivo de recurso |
|
Normalização de texto |
Arquivo de stop words |
|
|
Feature de tokenização de texto |
Arquivo de configuração de vocabulário |
|
|
Feature de relevância de texto |
Arquivo de configuração de frequência de termos |
|
|
Operador personalizado |
Arquivo de configuração do operador |
Criar a tabela de saída
No DataWorks, crie um nó PyOdps3 e execute o script a seguir. O script lê o arquivo fg.json, cria a tabela de saída e os recursos necessários para execuções subsequentes.
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)Antes de executar o script, configure os seguintes parâmetros em Scheduling Configuration: input_table, output_table, fg_json_file e partition_value.
Embora o método fg_task.run(o) também crie a tabela de saída automaticamente caso ela não exista, recomendamos chamar este método para criar a tabela antecipadamente. Isso ajuda a evitar conflitos e falhas de tarefa ao realizar backfill de dados simultaneamente.
Executar a tarefa offline do FG
No DataWorks, crie um nó PyOdps3 e execute o script a seguir. Este script executa a tarefa de geração de features e cria automaticamente a tabela de saída, caso ela ainda não exista.
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)
Antes de executar o script, configure os seguintes parâmetros em Scheduling Configuration: input_table, output_table, fg_json_file e partition_value.
Caso tenha o PyODPS instalado, você também pode instalar o pyfg e enviar tarefas localmente.
Parâmetros
|
Parâmetro |
Padrão |
Descrição |
|
input_table |
None |
A tabela de entrada. |
|
output_table |
None |
A tabela de saída. Criada automaticamente se não existir. |
|
fg_json_file |
None |
O arquivo de configuração do FG, no formato JSON. |
|
partition_value |
None |
A partição da tabela de entrada a ser processada. Os resultados são gravados na partição correspondente da tabela de saída. |
|
schema |
None |
O schema do MaxCompute. Para mais informações, consulte operações de schema. |
|
batch_size |
128 |
O número de registros a serem processados em cada lote. |
|
memory |
1024 |
A quantidade de memória a ser alocada para o nó da tarefa, em MiB. |
|
force_delete_output_table |
False |
Se definido como True, o sistema exclui a tabela de saída antes de executar a tarefa. |
|
force_update_resource |
False |
Se definido como True, o sistema atualiza os recursos antes de executar a tarefa. Para evitar conflitos de concorrência, evite manter este parâmetro definido como True. |
|
output_merged_str |
False |
Se definido como True, as strings são mescladas automaticamente para gerar uma feature de string grande no formato RTP. |
|
debug |
False |
Se definido como True, a tarefa é executada em modo de depuração e imprime o conteúdo de todos os recursos atualizados. |
|
sql_setting |
None |
Parâmetros SQL do MaxCompute. Utilize o método |
|
fg_setting |
None |
Parâmetros do FG. Utilize o método |
É possível substituir os valores padrão dos parâmetros passando-os ao construtor FgTask, conforme demonstrado nos exemplos.
Como funciona
O pacote pyfg é executado em uma máquina gateway dentro de um grupo de recursos exclusivo para enviar tarefas SQL ao MaxCompute. Alternativamente, instale tanto o pyfg quanto o PyODPS em uma máquina local para enviar tarefas.
UDFs personalizadas em uma tarefa SQL requerem recursos como a biblioteca compartilhada do FG, arquivos de configuração (por exemplo, fg.json, dicionários e bibliotecas de operadores personalizados) e os arquivos de código da UDF (.py). Todos esses recursos devem ser carregados no cluster do MaxCompute e armazenados em seu sistema de arquivos distribuído. Durante a execução da tarefa, cada worker baixa os recursos necessários do sistema de arquivos distribuído e os carrega na memória.
Alguns recursos, como a biblioteca compartilhada do FG e os arquivos de código da UDF, são compartilhados entre várias tarefas. Quando force_update_resource=True, o sistema exclui os recursos originais antes de carregar os novos. Esse processo cria uma janela de tempo que pode interromper outras tarefas em execução.