MaxCompute:Desenvolver uma tarefa PyODPS 2

Pré-requisitos

Antes de começar, verifique se você possui:

• Um nó PyODPS 2 criado. Consulte Criar e gerenciar nós do MaxCompute

Como funciona

O PyODPS é o SDK Python para o MaxCompute. No DataWorks, cada nó PyODPS inclui uma variável global pré-injetada odps (também referenciada como o) que serve como ponto de entrada do MaxCompute. Você escreve código Python usando esse ponto de entrada para consultar tabelas, executar instruções SQL, gerenciar recursos e processar dados. O DataWorks também injeta um dicionário global args para que seu código possa ler parâmetros de agendamento em tempo de execução.

A única diferença entre os nós PyODPS 2 e PyODPS 3 reside na camada Python: o PyODPS 2 utiliza Python 2.7, enquanto o PyODPS 3 utiliza Python 3.

Limites

O limite de memória de 50 MB aplica-se apenas a operações locais de dados. Tarefas SQL e DataFrame (exceto to_pandas ) iniciadas pelo PyODPS não estão sujeitas a essa restrição.

Observações de uso

Pacotes de terceiros

Para utilizar um pacote de terceiros em um nó PyODPS, use um grupo de recursos serverless e crie uma imagem personalizada que inclua o pacote.

Caso seu código UDF necessite de um pacote de terceiros, a abordagem de imagem personalizada não se aplica. Consulte Exemplo: Referenciar pacotes de terceiros em UDFs Python .

Acesso à rede

Para acessar uma source de dados em uma VPC ou em um data center on-premises, execute o nó em um grupo de recursos serverless e estabeleça uma conexão de rede entre o grupo de recursos e a source de dados. Consulte Soluções de conectividade de rede.

Atualização do PyODPS

• Em um grupo de recursos serverless: utilize o recurso de gerenciamento de imagens para executar /home/tops/bin/pip3 install pyodps==0.12.1 em um nó PyODPS 3 (substitua 0.12.1 pela versão desejada). Consulte Gerenciar imagens.

• Em um grupo de recursos exclusivo para agendamento: utilize o recurso O&M Assistant para executar o mesmo comando em um nó PyODPS 3. Consulte Usar o recurso O&M Assistant.

Logs de saída

Mantenha a saída de logs enxuta. Inclua logs de alerta e pontos de verificação de progresso em vez de despejar grandes conjuntos de dados no log. O limite de 4 MB aplica-se a todo o log de saída de uma execução de nó.

Linhagem de dados

Se as instruções SQL em um nó PyODPS não gerarem linhagens de dados no Data Map, passe os parâmetros de tempo de execução do agendador do DataWorks como hints SQL. O código a seguir demonstra como coletar esses parâmetros e passá-los durante a execução do SQL:

import os

# Collect DataWorks scheduler runtime parameters
skynet_hints = {}
for k, v in os.environ.items():
    if k.startswith('SKYNET_'):
        skynet_hints[k] = v

# Pass the parameters as hints when running SQL
o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)

Para mais informações, consulte Visualizar linhagens de dados e Configurar o parâmetro hints.

Escrever e executar código

Para obter a referência completa da sintaxe PyODPS, consulte Visão geral.

print(odps.exist_table('PyODPS_iris'))

options.tunnel.use_instance_tunnel = True
options.tunnel.limit_instance_tunnel = False  # Remove the record count limit

with instance.open_reader() as reader:
    # Reads all records using InstanceTunnel

with instance.open_reader(tunnel=True, limit=False) as reader:
    # Reads all records for this operation only

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

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

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

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

from odps.df import DataFrame
iris = DataFrame(o.get_table('pyodps_iris'))
for record in iris[iris.sepal_width < 3].execute():  # Triggers execution
    ...

from odps import options
from odps.df import DataFrame
options.interactive = True  # Enable at the top of the node
iris = DataFrame(o.get_table('pyodps_iris'))
print(iris.sepal_width.sum())  # Runs immediately and prints the result

Configurar parâmetros de agendamento

Diferentemente dos nós SQL, o código dos nós PyODPS não realiza substituição de strings ${param_name}. Em vez disso, o DataWorks injeta os parâmetros de agendamento no nó como um dicionário global args antes da execução.

Etapa 1: Adicionar parâmetros nas propriedades do nó

Abra a aba Properties do nó (painel de navegação à direita na aba de configuração) e adicione entradas na seção Scheduling Parameter. Para diferenças de formato e sintaxe entre tipos de nós, consulte Configurar parâmetros de agendamento para diferentes tipos de nós.

Etapa 2: Ler parâmetros no código

Por exemplo, se você definir ds=${yyyymmdd} na seção Scheduling Parameter, leia o valor no seu código da seguinte forma:

print('ds=' + args['ds'])
# Output: ds=20161116

Para obter a partição correspondente a essa data:

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

Os parâmetros de agendamento personalizados para nós PyODPS devem ser definidos com um valor constante. Ao contrário dos nós SQL, o valor não é substituído automaticamente em tempo de execução.

Para mais cenários de desenvolvimento de tarefas PyODPS, consulte:

• Referenciar um pacote de terceiros em um nó PyODPS

• Usar um nó PyODPS para segmentar texto chinês com base no Jieba

• Usar um nó PyODPS para baixar dados para um diretório local para processamento ou para processar dados online

Próximos passos

• Verificar se o nó foi executado com sucesso: A abordagem para confirmar a execução bem-sucedida de um script Shell também se aplica a scripts Python.

• Implantar o nó PyODPS 3: Em workspaces no modo padrão, implante o nó no ambiente de produção antes de agendá-lo.

• Realizar O&M no nó PyODPS 3: Após a implantação no Operation Center, gerencie e monitore o nó a partir do ambiente de produção.

• Perguntas frequentes sobre PyODPS: Problemas comuns e orientações de solução de problemas.