Exemplos de uso do SDK para Python: instruções SQL - MaxCompute: Central de Documentação do Alibaba Cloud

Observações de uso

Utilize execute_sql('statement') para execução síncrona e run_sql('statement') para execução assíncrona. Ambos os métodos retornam uma instância de tarefa em execução.
O MaxCompute não oferece suporte à leitura de resultados de instância no formato Arrow.
Algumas instruções SQL que funcionam no console do MaxCompute não podem ser executadas no PyODPS. Para instruções que não sejam DDL (Data Definition Language) ou DML (Data Manipulation Language), utilize o método apropriado:
- run_security_query — executa instruções GRANT e REVOKE
- run_xflow ou execute_xflow — executa instruções da Machine Learning Platform for AI (PAI)
Os jobs SQL são faturados com base na quantidade de jobs enviados. Consulte Visão geral do faturamento para obter detalhes.

Executar instruções SQL

Tanto execute_sql() quanto run_sql() enviam SQL para o MaxCompute. A diferença reside no fato de o seu código aguardar ou não pelo resultado.

import os
from odps import ODPS

# Store credentials in environment variables rather than hardcoding them
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='<your-endpoint>',
)

# Synchronous: blocks until the SQL job completes
o.execute_sql('select * from table_name')

# Asynchronous: returns immediately; the job runs in the background
instance = o.run_sql('select * from table_name')
print(instance.get_logview_address())  # Get the Logview URL for monitoring
instance.wait_for_success()            # Block until the job finishes

O método síncrono bloqueia a thread atual até que o job seja concluído. Já o método assíncrono retorna uma instância de tarefa imediatamente, permitindo que você envie vários jobs em paralelo e aguarde por eles seletivamente.

Ao executar consultas assíncronas em paralelo, certifique-se de compreender a ordem de dependência entre as consultas antes de enviá-las. Isole tarefas assíncronas de longa duração em conexões separadas quando elas precisarem sobreviver à sessão atual.

Definir parâmetros de tempo de execução

Passe um parâmetro hints como um dict para substituir as configurações da sessão em uma única execução:

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

Para aplicar as mesmas configurações a todas as execuções da sessão, configure options.sql.settings globalmente:

from odps import options

options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # Picks up the global hints automatically

Ler resultados da consulta

Utilize open_reader para acessar os resultados após a conclusão de uma consulta. Escolha a abordagem com base no retorno da consulta e no volume de dados necessário.

Cenário	Abordagem
Dados estruturados (conjunto de resultados pequeno a médio)	Itere sobre registros com `open_reader()`
Saída de texto bruto (por exemplo, comando `DESC`)	Leia `reader.raw`
Carregar todos os resultados na memória	`reader.to_pandas()`
Conjunto de resultados grande com download paralelo	`reader.to_pandas(n_process=N)`

Dados estruturados — itere sobre os registros:

with o.execute_sql('select * from table_name').open_reader() as reader:
    for record in reader:
        print(record)

Saída bruta (por exemplo, de um comando DESC) — leia reader.raw:

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

Escolher entre Instance Tunnel e a interface Result

O método open_reader suporta dois backends:

Backend	Quando usar	Como selecionar
Instance Tunnel	Opcional; suporta grandes conjuntos de resultados	Defina `options.tunnel.use_instance_tunnel = True` e use `tunnel=True`
Interface Result	Padrão, ou fallback para versões antigas do MaxCompute ou em caso de erros no Instance Tunnel	`tunnel=False` (ou defina `options.tunnel.use_instance_tunnel = False`)

O PyODPS faz downgrade automaticamente para a interface Result se o Instance Tunnel falhar e registra um alerta explicando o motivo. Para alternar os backends manualmente:

# Use Instance Tunnel explicitly
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    for record in reader:
        print(record)

# Use the Result interface explicitly
with o.execute_sql('select * from dual').open_reader(tunnel=False) as reader:
    for record in reader:
        print(record)

Limitar o número de registros baixados

Para limitar o número de registros retornados, passe uma opção limit para open_reader ou defina options.tunnel.limit_instance_tunnel = True. Sem uma configuração explícita, o MaxCompute aplica o limite de Tunnel no nível do projeto — geralmente 10.000 registros por download.

Ler resultados em um DataFrame do pandas

Chame to_pandas() no reader para carregar todos os resultados em um DataFrame do pandas:

with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    pd_df = reader.to_pandas()  # Returns a pandas DataFrame

Acelerar a leitura com múltiplos processos

A leitura multiprocessos requer PyODPS 0.11.3 ou posterior.

Passe n_process para to_pandas() para distribuir o download entre vários processos. O SDK divide o conjunto de resultados em lotes e realiza o download deles em paralelo.

import multiprocessing

n_process = multiprocessing.cpu_count()
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    pd_df = reader.to_pandas(n_process=n_process)

Configurar aliases de recursos para UDFs

Quando o SQL executa uma função definida pelo usuário (UDF) que referencia um arquivo de recurso e o conteúdo desse recurso é alterado, normalmente seria necessário excluir e recriar a UDF. Em vez disso, utilize o parâmetro aliases para mapear o nome antigo do recurso para o novo — sem necessidade de alterar a UDF.

O exemplo a seguir configura uma UDF que lê um inteiro de um arquivo de recurso e o soma ao valor de entrada:

from odps.models import Schema

myfunc = '''\
from odps.udf import annotate
from odps.distcache import get_cache_file

@annotate('bigint->bigint')
class Example(object):
    def __init__(self):
        self.n = int(get_cache_file('test_alias_res1').read())

    def evaluate(self, arg):
        return arg + self.n
'''

res1 = o.create_resource('test_alias_res1', 'file', file_obj='1')
o.create_resource('test_alias.py', 'py', file_obj=myfunc)
o.create_function('test_alias_func',
                  class_type='test_alias.Example',
                  resources=['test_alias.py', 'test_alias_res1'])

table = o.create_table(
    'test_table',
    schema=Schema.from_lists(['size'], ['bigint']),
    if_not_exists=True
)

data = [[1, ], ]
o.write_table(table, 0, [table.new_record(it) for it in data])

with o.execute_sql(
    'select test_alias_func(size) from test_table').open_reader() as reader:
    print(reader[0][0])

Para executar a mesma consulta com um recurso diferente — sem modificar a UDF — passe aliases:

res2 = o.create_resource('test_alias_res2', 'file', file_obj='2')

# Map the old resource name to the new one at query time
with o.execute_sql(
    'select test_alias_func(size) from test_table',
    aliases={'test_alias_res1': 'test_alias_res2'}).open_reader() as reader:
    print(reader[0][0])

Executar SQL em ambientes interativos

O PyODPS inclui plug-ins SQL para IPython e Jupyter que suportam consultas parametrizadas. Consulte a documentação de aprimoramento da experiência do usuário para obter instruções de configuração.

Definir biz_id

Em alguns casos, pode ser necessário enviar o biz_id ao submeter instruções SQL. Caso contrário, ocorrerá um erro durante a execução das instruções. Defina-o globalmente para que se aplique a todas as execuções na sessão:

from odps import options

options.biz_id = 'my_biz_id'
o.execute_sql('select * from pyodps_iris')