DataWorks:Nó PyODPS 2

Visão geral

O PyODPS é o SDK Python do MaxCompute. Ele fornece uma interface de programação concisa que permite usar Python para escrever jobs, consultar tabelas e visualizações, além de gerenciar recursos do MaxCompute. Para mais informações, consulte PyODPS overview. No DataWorks, use um nó PyODPS para agendar e executar tarefas Python e integrá-las a outros jobs.

Observações

• Se seu código exigir pacotes de terceiros ao executar um nó PyODPS em um grupo de recursos do DataWorks, utilize um serverless resource group e uma custom image para instalar os pacotes.

  Nota

  Esse método não é compatível se o código usar uma UDF que referencie um pacote de terceiros. Para saber como lidar com esse cenário, consulte UDF example: Use third-party packages in Python UDFs.

• Se a tarefa PyODPS precisar acessar um ambiente de rede específico, como uma fonte de dados ou service em uma VPC ou data center on-premises, utilize um grupo de recursos serverless e consulte Network connectivity solutions para estabelecer a conexão de rede necessária.

• Para obter mais detalhes sobre a sintaxe do PyODPS, consulte a PyODPS documentation.

• Os nós PyODPS estão disponíveis em dois tipos: PyODPS 2 e PyODPS 3, diferenciando-se pela versão subjacente do Python. Nós PyODPS 2 utilizam Python 2, enquanto nós PyODPS 3 utilizam Python 3. Crie o tipo de nó correspondente à versão do Python utilizada.

• Se a execução de SQL em um nó PyODPS falhar ao gerar linhagem de dados, impedindo sua exibição no Data Map, resolva esse problema definindo manualmente os parâmetros de agendamento do DataWorks no código da tarefa. Para visualizar a linhagem de dados, consulte View data lineage. Para informações sobre configurações de parâmetros, consulte Set runtime parameter hints. Obtenha os parâmetros de tempo de execução necessários usando o seguinte código de exemplo.

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

• O log de saída de um nó PyODPS suporta tamanho máximo de 4 MB. Evite imprimir grandes volumes de dados diretamente no log. Registre apenas avisos essenciais e atualizações de progresso.

Limitações

• Ao executar um nó PyODPS em um grupo de recursos exclusivo para agendamento, recomenda-se que os dados processados localmente dentro do nó não ultrapassem 50 MB. Essa operação é limitada pelas especificações do grupo de recursos exclusivo para agendamento. Processar muitos dados locais que excedam o limiar do sistema operacional pode causar um erro OOM (got killed). Evite escrever excesso de código de processamento de dados em nós PyODPS.

• Ao executar um nó PyODPS em um grupo de recursos serverless, configure a alocação de CU para o nó PyODPS com base no volume de dados a ser processado.

  Nota

  Ao executar esta tarefa em um grupo de recursos serverless, a configuração máxima para uma única tarefa é 64CU, mas recomenda-se não exceder 16CU para evitar insuficiência de recursos causada por alocação excessiva de CU, o que pode afetar a inicialização da tarefa.

• Caso encontre um erro Got Killed, isso indica que o uso de memória excedeu o limite e o processo foi encerrado. Portanto, evite operações de dados locais sempre que possível. Tarefas SQL e DataFrame iniciadas via PyODPS (exceto to_pandas) não estão sujeitas a essa limitação.

• Códigos não-UDF podem usar os pacotes NumPy e Pandas pré-instalados. Outros pacotes de terceiros que contenham código binário não são compatíveis.

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

• A versão subjacente do Python para nós PyODPS 2 é a 2,7.

• Não há suporte para executar várias tarefas Python simultaneamente dentro de um nó PyODPS.

• Para imprimir logs em um nó PyODPS, utilize print. O uso de logger.info não é compatível.

Pré-requisitos

Associate a MaxCompute compute engine with your DataWorks workspace.

Procedimento

1. Na página de edição do nó PyODPS 2, realize as seguintes operações de desenvolvimento.

   Exemplos de código PyODPS 2

   Após criar um nó PyODPS, edite e execute o código. Para mais informações sobre a sintaxe do PyODPS, consulte a PyODPS documentation. Este tópico descreve os cinco exemplos de código a seguir. Selecione os exemplos conforme suas necessidades de negócio.

   ODPS entry

   Em nós PyODPS do DataWorks, uma variável global odps ou o está incluída como ponto de entrada ODPS. Não é necessário definir manualmente o ponto de entrada ODPS.

   print(odps.exist_table('PyODPS_iris'))

   Execute SQL

   É possível executar SQL em um nó PyODPS. Para mais informações, consulte Execute SQL statements.

   • Por padrão, o instance tunnel não está ativado no DataWorks, o que significa que instance.open_reader usa a interface Result por padrão (até 10.000 registros). Utilize reader.count para obter o número de registros. Caso precise iterar sobre todos os dados, desative a restrição limit. Use as instruções a seguir para ativar o Instance Tunnel globalmente e desativar a restrição limit.

     options.tunnel.use_instance_tunnel = True
     options.tunnel.limit_instance_tunnel = False  # Disable limit restriction, read all data.
     
     with instance.open_reader() as reader:
       # Use Instance Tunnel to read all data.

   • Alternativamente, adicione tunnel=True ao open_reader para ativar o instance tunnel apenas na chamada atual do open_reader. Também é possível adicionar limit=False para desativar a restrição limit somente na chamada atual.

     # The open_reader uses the Instance Tunnel interface to read all data.
     with instance.open_reader(tunnel=True, limit=False) as reader:

   Set runtime parameters

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

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

   • Após configurar sql.settings globalmente, adicione os parâmetros de tempo de execução relevantes sempre que executar uma tarefa.

     from odps import options
     options.sql.settings = {'odps.sql.mapper.split.size': 16}
     o.execute_sql('select * from PyODPS_iris')  # Add hints based on global configurationhints。

   Read execution results

   Uma instância que executa SQL pode realizar diretamente a operação open_reader. Dois cenários são possíveis:

   • A instrução SQL retorna dados estruturados.

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

   • A instrução SQL pode ser algo como desc. Utilize a propriedade reader.raw para obter o resultado bruto da execução SQL.

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

     Nota

     Se utilizar parâmetros de agendamento personalizados, codifique rigidamente os valores de tempo ao acionar diretamente um nó PyODPS 2 na página, pois um nó PyODPS não pode substituir os parâmetros diretamente.

   DataFrame

   Também é possível usar DataFrame para processar dados.

   • Execução

     No ambiente DataWorks, a execução de DataFrame requer chamadas explícitas para methods that trigger immediate execution.

     from odps.df import DataFrame
     iris = DataFrame(o.get_table('pyodps_iris'))
     for record in iris[iris.sepal_width < 3].execute():  # Execute immediately and process each Record.

     Para acionar a execução imediata ao usar Print, ative options.interactive.

     from odps import options
     from odps.df import DataFrame
     options.interactive = True  # Enable the switch at the beginning.
     iris = DataFrame(o.get_table('pyodps_iris'))
     print(iris.sepal_width.sum())  # Executes immediately on print.

   • Imprimir informações detalhadas

     Configure a opção options.verbose. No DataWorks, essa opção vem ativada por padrão, e informações detalhadas, como a URL do Logview, são impressas durante a execução.

   Desenvolvimento de código PyODPS 2

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

   1. Prepare um conjunto de dados e crie a tabela de amostra pyodps_iris. Para mais informações, consulte Create a DataFrame.

   2. Crie um DataFrame. Para mais informações, consulte Create a DataFrame.

   3. Insira o seguinte código no nó PyODPS.

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

   Executar uma tarefa PyODPS

   1. Em Run Configuration, configure Compute Resource, Compute Resource and Quota e DataWorks Resource Group.

      Nota

      • Para acessar fontes de dados em ambiente de rede pública ou VPC, utilize um grupo de recursos de agendamento que tenha passado no teste de conectividade para a fonte de dados. Para mais informações, consulte Network connectivity solutions.

      • Configure as informações de Image conforme os requisitos da tarefa.

   2. Na barra de ferramentas, clique em Run para executar a tarefa PyODPS.

2. Caso precise executar a tarefa do nó periodicamente, configure as definições de agendamento conforme suas necessidades de negócio. Para mais informações, consulte Configure schedule settings.

   Diferentemente dos nós SQL no DataWorks, os nós PyODPS não substituem strings como ${param_name} no código para evitar impactos. Em vez disso, antes de executar o código, um dict chamado args é adicionado às variáveis globais, permitindo recuperar os parâmetros de agendamento. Por exemplo, se definir ds=${yyyymmdd} em Parameter, recupere o parâmetro no código da seguinte forma.

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

   Nota

   Para obter a partição chamada ds, utilize o método abaixo.

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

3. Após concluir a configuração da tarefa do nó, implante o nó. Para mais informações, consulte Deploy a node.

4. Depois de implantar a tarefa, visualize o status de execução da tarefa agendada no Operation Center. Para mais informações, consulte View scheduled tasks.

Executar um nó usando uma função associada

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

Próximo passo

PyODPS FAQ: Conheça problemas comuns durante a execução do PyODPS para solucionar falhas rapidamente.