MaxCompute:SQL

Parámetros

Valores devueltos

Tanto execute_sql() como run_sql() devuelven información de Task instances.

Ejemplos

Ejecución síncrona frente a asíncrona

# Synchronous: blocks until the statement finishes
o.execute_sql('select * from table_name')

# Asynchronous: returns immediately
instance = o.run_sql('select * from table_name')
print(instance.get_logview_address())  # Get the LogView URL
instance.wait_for_success()  # Block until the statement finishes

Pasar sugerencias de tiempo de ejecución

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

Configuración global

Establezca options.sql.settings para aplicar parámetros de tiempo de ejecución a cada llamada posterior de execute_sql(). Consulte Flag parameters para ver los parámetros disponibles.

from odps import options
options.sql.settings = {'odps.stage.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # Hints apply automatically

Datos estructurados (SELECT)

Las consultas SELECT devuelven registros estructurados. Use un bucle for para iterar sobre cada registro:

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

Datos no estructurados (DESC y otros comandos)

Los comandos como desc devuelven texto sin procesar. Acceda a este mediante reader.raw:

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

Interfaz InstanceTunnel frente a Result

De manera predeterminada, open_reader() usa la interfaz Result, la cual puede agotar el tiempo de espera o limitar la cantidad de registros devueltos. Habilite InstanceTunnel para lecturas completas de datos:

Opción 1: configuración global

from odps import options
options.tunnel.use_instance_tunnel = True

Opción 2: parámetro por llamada

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

En PyODPS V0.7.7.1 y versiones posteriores, open_reader() admite la lectura completa de datos con este enfoque.

• Si su versión de MaxCompute es anterior o InstanceTunnel genera un error, PyODPS emite una alerta y vuelve automáticamente a la interfaz Result. Revise el mensaje de alerta para identificar la causa.

• Si su versión de MaxCompute solo admite la interfaz Result y necesita todos los resultados, escríbalos primero en otra tabla. Después, lea esa tabla con open_reader(). Este método está sujeto a las restricciones de data protection mechanism de su proyecto.

• Para obtener más información sobre InstanceTunnel, consulte InstanceTunnel.

Modo de límite de lectura

De manera predeterminada, PyODPS no limita los datos que se leen de una instancia. Sin embargo, si el propietario del proyecto configuró protección de datos, PyODPS habilita automáticamente el modo de límite de lectura cuando detecta restricciones y options.tunnel.limit_instance_tunnel no está definido. En la mayoría de los casos, este modo restringe la lectura a 10 000 filas.

Habilitar manualmente el modo de límite de lectura (proyectos protegidos):

# Option A: per-call
reader = instance.open_reader(limit=True)

# Option B: global
options.tunnel.limit_instance_tunnel = True

Deshabilite el modo de límite de lectura para acceder a todos los datos:

with o.execute_sql('select * from table_name').open_reader(tunnel=True, limit=False) as reader:
    for record in reader:
        print(record)

En entornos como DataWorks, options.tunnel.limit_instance_tunnel puede tener el valor predeterminado True . Para leer todos los datos, pase tanto tunnel=True como limit=False a open_reader() .

Proyectos protegidos

Si su proyecto está protegido y tunnel=True, limit=False no elimina la restricción, comuníquese con el propietario del proyecto para que le otorgue permisos de lectura. Consulte Project data protection para obtener más detalles.

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, ], ]
# Write one row of data with value 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])

res2 = o.create_resource('test_alias_res2', 'file', file_obj='2')
# Map res1 alias to res2 without modifying the UDF or resource
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])

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