云原生大数据计算服务 MaxCompute：Python SDK示例：SQL

PyODPS支持MaxCompute SQL查询，并可以读取执行的结果，使用时有以下注意事项。

• 入口对象的execute_sql('statement')和run_sql('statement')方法可以执行SQL语句，返回值是运行实例，详情请参见任务实例。

• 目前暂不支持使用Arrow格式读取Instance结果。

• 并非所有可以执行的MaxCompute命令都是PyODPS可以接受的SQL语句。在调用非DDL、DML语句时，请使用其他方法，例如：

  • GRANT、REVOKE等语句请使用run_security_query方法。

  • PAI命令请使用run_xflow或execute_xflow方法。

• 调用SQL引擎执行SQL，会按照SQL作业进行计费，计费详情请参见计费项与计费方式概述。

在运行时如果需要设置参数，您可以通过设置hints参数来实现，参数的类型是dict。

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

您可以对于全局配置设置sql.settings，后续每次运行时则都会添加相关的运行时参数。

from odps import options
options.sql.settings = {'odps.sql.mapper.split.size': 16}
o.execute_sql('select * from pyodps_iris')  # 会根据全局配置添加hints

运行SQL的Instance能够直接执行open_reader操作读取SQL执行结果。读取时会出现以下两种情况：

• SQL返回了结构化的数据。

  with o.execute_sql('select * from table_name').open_reader() as reader:
      for record in reader:
          print(record)  # 处理每一个record

• SQL可能执行了desc命令，这时可以通过reader.raw取到原始的SQL执行结果。

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

如果您设置了options.tunnel.use_instance_tunnel == True，在后续调用open_reader时，PyODPS会默认调用Instance Tunnel， 否则会调用旧的Result接口。如果您使用了版本较低的 MaxCompute服务，或者调用Instance Tunnel出现了问题，PyODPS会给出告警并自动降级到旧的Result接口，您可根据告警信息判断导致降级的原因。如果Instance Tunnel的返回结果不合预期， 您可以将该选项设为False，在调用open_reader时，也可以使用tunnel参数来指定使用何种结果接口。

• 使用Instance Tunnel

  with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
      for record in reader:
          print(record)  # 处理每一个record

• 使用 Results 接口

  with o.execute_sql('select * from dual').open_reader(tunnel=False) as reader:
      for record in reader:
          print(record)  # 处理每一个record

您可以通过n_process指定使用进程数。

import multiprocessing
n_process = multiprocessing.cpu_count()
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    # n_process 指定成机器核数
    pd_df = reader.to_pandas(n_process=n_process)

在运行SQL时，如果某个UDF引用的资源是动态变化的，您可以alias旧的资源名到新的资源，这样就可以避免重新删除并重新创建UDF。

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, ], ]
# 写入一行数据，只包含一个值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')
# 把内容为1的资源alias成内容为2的资源，不需要修改UDF或资源
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])

在少数情形下，在提交SQL时，可能需要同时提交biz_id，否则执行会报错。此时，您可以在全局options里设置biz_id。

from odps import options

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