PyODPS支持MaxCompute SQL查詢，并可以讀取執(zhí)行的結(jié)果，使用時有以下注意事項。

• 入口對象的execute_sql('statement')和run_sql('statement')方法可以執(zhí)行SQL語句，返回值是運(yùn)行實例，詳情請參見任務(wù)實例。

• 目前暫不支持使用Arrow格式讀取Instance結(jié)果。

• 并非所有可以執(zhí)行的MaxCompute命令都是PyODPS可以接受的SQL語句。在調(diào)用非DDL、DML語句時，請使用其他方法，例如：

  • GRANT、REVOKE等語句請使用run_security_query方法。

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

• 調(diào)用SQL引擎執(zhí)行SQL，會按照SQL作業(yè)進(jìn)行計費(fèi)，計費(fèi)詳情請參見計費(fèi)項與計費(fèi)方式概述。

在運(yùn)行時如果需要設(shè)置參數(shù)，您可以通過設(shè)置hints參數(shù)來實現(xiàn)，參數(shù)的類型是dict。

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

您可以對于全局配置設(shè)置sql.settings，后續(xù)每次運(yùn)行時則都會添加相關(guān)的運(yùn)行時參數(shù)。

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

運(yùn)行SQL的Instance能夠直接執(zhí)行open_reader操作讀取SQL執(zhí)行結(jié)果。讀取時會出現(xiàn)以下兩種情況：

• SQL返回了結(jié)構(gòu)化的數(shù)據(jù)。

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

• SQL可能執(zhí)行了desc命令，這時可以通過reader.raw取到原始的SQL執(zhí)行結(jié)果。

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

如果您設(shè)置了options.tunnel.use_instance_tunnel == True，在后續(xù)調(diào)用open_reader時，PyODPS會默認(rèn)調(diào)用Instance Tunnel， 否則會調(diào)用舊的Result接口。如果您使用了版本較低的 MaxCompute服務(wù)，或者調(diào)用Instance Tunnel出現(xiàn)了問題，PyODPS會給出告警并自動降級到舊的Result接口，您可根據(jù)告警信息判斷導(dǎo)致降級的原因。如果Instance Tunnel的返回結(jié)果不合預(yù)期， 您可以將該選項設(shè)為False，在調(diào)用open_reader時，也可以使用tunnel參數(shù)來指定使用何種結(jié)果接口。

• 使用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指定使用進(jìn)程數(shù)。

import multiprocessing
n_process = multiprocessing.cpu_count()
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
    # n_process 指定成機(jī)器核數(shù)
    pd_df = reader.to_pandas(n_process=n_process)

在運(yùn)行SQL時，如果某個UDF引用的資源是動態(tài)變化的，您可以alias舊的資源名到新的資源，這樣就可以避免重新刪除并重新創(chuàng)建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, ], ]
# 寫入一行數(shù)據(jù)，只包含一個值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')
# 把內(nèi)容為1的資源alias成內(nèi)容為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])

在少數(shù)情形下，在提交SQL時，可能需要同時提交biz_id，否則執(zhí)行會報錯。此時，您可以在全局options里設(shè)置biz_id。

from odps import options

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