DataWorks:Nó PyODPS 3

Introdução

O PyODPS é o SDK Python para o MaxCompute. Ele permite escrever jobs, consultar tabelas e visualizações, além de gerenciar recursos do MaxCompute com Python. Para mais informações, consulte PyODPS overview. No DataWorks, os nós PyODPS permitem agendar e executar tarefas Python juntamente com outros jobs.

Observações de uso

• Para chamar um pacote de terceiros em um nó PyODPS dentro de um grupo de recursos do DataWorks, use um serverless resource group com uma custom image.

  Nota

  Este método não se aplica caso seu código inclua uma função definida pelo usuário (UDF) que referencie um pacote de terceiros. Para o procedimento correto, consulte UDF example: Use a third-party package in a Python UDF.

• Para atualizar a versão do PyODPS, use uma custom image em um grupo de recursos serverless para executar o comando /home/tops/bin/pip3 install pyodps==0.12.1. Substitua 0.12.1 pela versão desejada do PyODPS. Em grupos de recursos exclusivos para agendamento, use o O&M Assistant para executar o mesmo comando.

• Caso seu job PyODPS precise acessar um ambiente de rede específico, como uma fonte de dados ou service em uma VPC ou rede IDC, use um grupo de recursos serverless. Para mais detalhes sobre como conectar o grupo de recursos serverless ao ambiente de destino, consulte Network connectivity solutions.

• Para mais informações sobre a sintaxe do PyODPS, consulte a PyODPS documentation.

• Há dois tipos de nós PyODPS: PyODPS 2 (Python 2) e PyODPS 3 (Python 3). Crie o tipo de nó correspondente à sua versão do Python.

• Se a execução de instruções SQL em um nó PyODPS não gerar a linhagem de dados correta no Data Map, defina manualmente os parâmetros de agendamento relevantes do DataWorks no código do job. Para saber como visualizar a linhagem de dados, consulte View data lineage. Para obter detalhes sobre como definir parâmetros, consulte Set runtime parameter hints. O exemplo de código a seguir obtém os parâmetros necessários durante a execução.

  import os
  ...
  # get DataWorks scheduler runtime parameters
  skynet_hints = {}
  for k, v in os.environ.items():
      if k.startswith('SKYNET_'):
          skynet_hints[k] = v
  ...
  # setting hints while submitting a job
  o.execute_sql('INSERT OVERWRITE TABLE XXXX SELECT * FROM YYYY WHERE ***', hints=skynet_hints)
  ...

• O log de saída de um nó PyODPS tem tamanho máximo de 4 MB. Evite imprimir grandes volumes de dados no log. Concentre-se na emissão de informações de alerta e progresso.

Limitações

• Ao executar um nó PyODPS em um grupo de recursos exclusivo para agendamento, recomenda-se que os dados processados localmente não ultrapassem 50 MB. Esse limite decorre das especificações do grupo de recursos exclusivo. Se os dados locais excederem o limiar do sistema operacional, poderá ocorrer um erro OOM (Got Killed). Evite escrever código excessivo de processamento de dados no nó PyODPS.

• Ao executar um nó PyODPS em um grupo de recursos serverless, configure uma quantidade adequada de CUs conforme o volume de dados.

  Nota

  Em um grupo de recursos serverless, um único job suporta no máximo 64 CUs. Contudo, recomenda-se não ultrapassar 16 CUs para evitar escassez de recursos que possa impedir o início do job.

• Um erro Got killed indica que o processo excedeu o limite de memória. Evite operações de dados locais. Jobs SQL e DataFrame (exceto operações to_pandas) iniciados via PyODPS não estão sujeitos a essa limitação.

• Use as bibliotecas NumPy e pandas pré-instaladas em códigos que não envolvam funções definidas pelo usuário. Outros pacotes de terceiros contendo código binário não são suportados.

• Por motivos de compatibilidade, options.tunnel.use_instance_tunnel vem definido como False por padrão no DataWorks. Para ativar o instance tunnel globalmente, defina manualmente esse valor como True.

• A definição de bytecode varia entre versões menores do Python 3, como Python 3.8 e Python 3.7.

  Atualmente, o MaxCompute utiliza Python 3.7. Caso você use sintaxe de outras versões do Python 3, como o finally block do Python 3.8, ocorrerá um erro durante a execução. Recomendamos o uso do Python 3.7.

• O PyODPS 3 suporta execução em grupo de recursos serverless. Para adquirir e usar um, consulte Use a serverless resource group.

• Não é possível configurar múltiplos jobs Python para execução simultânea dentro de um único nó PyODPS.

• Para imprimir logs em um nó PyODPS, use print. O uso de logger.info não é suportado.

Pré-requisitos

Associate a MaxCompute compute engine with your DataWorks workspace.

Procedimento

1. Desenvolva seu código na página do editor do nó PyODPS 3.

   Exemplos de código PyODPS 3

   Após criar um nó PyODPS, edite e execute seu código. Para mais informações sobre a sintaxe do PyODPS, consulte Basic operations. Os exemplos abaixo abrangem cinco cenários comuns. Escolha aquele que melhor atende às suas necessidades.

   ODPS entry point

   Cada nó PyODPS no DataWorks inclui uma variável global de ponto de entrada ODPS, odps ou o, que não precisa ser definida manualmente.

   print(odps.exist_table('PyODPS_iris'))

   Execute SQL

   Execute instruções SQL em um nó PyODPS. Para mais detalhes, consulte SQL.

   • Por padrão, o instance tunnel vem desativado no DataWorks, fazendo com que instance.open_reader utilize a interface Result e retorne no máximo 10.000 registros. Use reader.count para obter a quantidade de registros. Para iterar por todos os dados, desative o limit. As instruções a seguir ativam globalmente o instance tunnel e desativam o limit.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable the limit to read all data.
     
     with instance.open_reader() as reader:
       # All data can be read through Instance Tunnel.

   • Ative o instance tunnel para uma única chamada de open_reader adicionando tunnel=True à chamada de open_reader. Adicione também limit=False para remover a restrição de limit nessa chamada específica.

     # Use Instance Tunnel for this open_reader call and read all data.
     with instance.open_reader(tunnel=True, limit=False) as reader:

   Set runtime parameters

   • Defina parâmetros de execução usando o parâmetro hints, que é um dict. Para mais informações sobre hints, consulte SET operations.

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

   • Ao definir configurações globais via sql.settings, os parâmetros de execução correspondentes serão adicionados a cada execução.

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

   Read execution results

   Chame open_reader diretamente em uma instância de execução SQL. Isso abrange dois cenários:

   • A instrução SQL retorna dados estruturados.

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

   • Ao executar instruções SQL como desc, use a propriedade reader.raw para obter os resultados brutos da execução SQL.

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

     Nota

     Se você usar parâmetros de agendamento personalizados e executar um nó PyODPS 3 diretamente pela UI, será necessário codificar o valor de tempo fixo, pois o nó não substitui variáveis em tempo de execução.

   DataFrame

   Processe dados também usando DataFrame (Deprecated).

   • Execução

     No ambiente DataWorks, as operações de DataFrame exigem uma chamada explícita a um immediate execution method.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Call an immediate execution method to process each record.

     Caso precise acionar a execução imediata ao imprimir, ative options.interactive.

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # Enable the option at the beginning.
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # This triggers immediate execution.

   • Impressão de informações detalhadas

     A opção options.verbose vem ativada por padrão no DataWorks, fazendo com que informações detalhadas, como a URL do Logview, sejam impressas durante a execução.

   Desenvolvimento de código PyODPS 3

   O exemplo a seguir demonstra como usar um nó PyODPS:

   1. Prepare um conjunto de dados criando a tabela de amostra pyodps_iris. Para mais informações, consulte Process DataFrame data.

   2. Crie um DataFrame. Para mais detalhes, consulte Create a DataFrame from a MaxCompute table.

   3. Insira o código abaixo no nó PyODPS e execute-o.

      from odps.df import DataFrame
      
      # Create a DataFrame from an ODPS table.
      iris = DataFrame(o.get_table('pyodps_iris'))
      print(iris.sepallength.head(5))

   Executar o job PyODPS****

   1. No painel Run Configuration, na seção Compute Resource, configure o Compute Resource, a computing quota e o DataWorks Resource Group.

      Nota

      • Para acessar uma fonte de dados via rede pública ou VPC, use um grupo de recursos de agendamento que tenha passado no teste de conectividade com a fonte de dados. Para mais informações, consulte Network connectivity solutions.

      • Configure as informações de Image conforme os requisitos do job.

   2. Na caixa de diálogo de parâmetros da barra de ferramentas, selecione a fonte de dados MaxCompute criada e clique em Run para executar o job PyODPS.

2. Para executar o nó periodicamente, configure suas propriedades de agendamento conforme seus requisitos de negócio. Para mais informações, consulte Configure node scheduling.

   Diferentemente dos nós SQL no DataWorks, os nós PyODPS não substituem strings como ${param_name} no código. Em vez disso, antes da execução do código, um dicionário chamado args é adicionado às variáveis globais, permitindo recuperar os parâmetros de agendamento. Por exemplo, se você definir ds=${yyyymmdd} na seção Parameter, recupere a informação do parâmetro no código da seguinte forma.

   print('ds=' + args['ds'])
   ds=20240930

   Nota

   Caso precise obter a partição chamada ds, use o método abaixo.

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

3. Após configurar o nó, faça o deploy dele. Para mais informações, consulte Node and workflow deployment.

4. Depois de fazer o deploy do job, visualize seu status no Operation Center. Para mais detalhes, consulte Get started with Operation Center.

Executar um nó usando uma função associada

Você pode associate a RAM role to run a node, o que permite executar tarefas de nó com uma função RAM específica para controle de permissões granular e gerenciamento de segurança.

Próximos passos

FAQ about PyODPS: Problemas comuns durante a execução do PyODPS e como resolvê-los.