本文為您介紹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作業進行計費,計費詳情請參見計費項目與計費方式概述。
執行SQL
import os
from odps import ODPS
# 確保 ALIBABA_CLOUD_ACCESS_KEY_ID 環境變數設定為使用者 Access Key ID,
# ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境變數設定為使用者 Access Key Secret,
# 不建議直接使用 Access Key ID / Access Key Secret 字串
o = ODPS(
os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
project='your-default-project',
endpoint='your-end-point',
)
o.execute_sql('select * from table_name') # 同步的方式執行,會阻塞直到SQL執行完成。
instance = o.run_sql('select * from table_name') # 以非同步方式提交
print(instance.get_logview_address()) # 擷取logview地址
instance.wait_for_success() # 阻塞直到完成
設定運行參數
在運行時如果需要設定參數,您可以通過設定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執行結果
運行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
設定讀取資料規模
如果您想要限制下載資料的規模,可以為open_reader
增加limit選項, 或者設定 options.tunnel.limit_instance_tunnel = True
。如果未設定 options.tunnel.limit_instance_tunnel
,MaxCompute會自動開啟資料量限制,此時,可下載的資料條數受到Project配置的Tunnel下載資料規模數限制, 通常該限制為10000條。
設定讀取結果為pandas DataFrame
# 直接使用 reader 的 to_pandas 方法
with o.execute_sql('select * from dual').open_reader(tunnel=True) as reader:
# pd_df 類型為 pandas DataFrame
pd_df = reader.to_pandas()
設定讀取速度(進程數)
多進程加速僅在 PyODPS 0.11.3 及以上版本中支援。
您可以通過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)
設定alias
在運行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
在ipython和jupyter裡支援使用SQL外掛程式的方式運行SQL,且支援參數化查詢, 詳情參考互動體驗增強文檔。
設定biz_id
在少數情形下,在提交SQL時,可能需要同時提交biz_id,否則執行會報錯。此時,您可以在全域options裡設定biz_id。
from odps import options
options.biz_id = 'my_biz_id'
o.execute_sql('select * from pyodps_iris')