PyODPS是MaxCompute的Python SDK,提供DataFrame框架和MaxCompute对象的基本操作方法。
接口手册:https://pyodps.readthedocs.io/zh_CN/latest/?spm=a2c4g.11186623.0.0.1aaf3d94n84mIN
PySdk下载:https://github.com/aliyun/aliyun-odps-python-sdk
pip install ODPS
确认下载好后,连接odps,账号密码自备。
from odps import ODPS
o = ODPS(
access_id='user', #登陆账号
secret_access_key='password', #登陆密码
project='project', #odps上的项目名称
endpoint='http://service.cn-hangzhou-xxx:80/api') #官方提供的接口
3.1、简单执行sql
with o.execute_sql('desc tablename').open_reader() as reader:
print(reader.raw)
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() # 阻塞直到完成。
3.2、通过odps内置DataFrame读取,该方法读取的数据结构类型为odps.df.expr.core.DataFrame
def get_odps_table(tb_name):
data = DataFrame(o.get_table(tb_name))
data['ds'] = data['ds'].astype('int')
return data
rdata = get_odps_table('tb_name') #获取表数据实例
3.3、封装成函数连接
直接输入sql就可进行增删改查
def exe_sql(sql):
from datetime import datetime
import pandas as pd
st_time = datetime.now()
data = []
with o.execute_sql(sql).open_reader() as reader:
d = defaultdict(list) # collection默认一个dict
# 解析record中的每一个元组,存储方式为(k,v),以k作为key,存储每一列的内容;
for record in reader:
for res in record:
d[res[0]].append(res[1])
data = pd.DataFrame.from_dict(d, orient='index').T # 转换为数据框,并转置,不转置的话是横条数据
ed_time = datetime.now()
print('总耗时:', ed_time - st_time ) # 输出sql运行时间,方便sql修改、维护
return data
3.4、将sql结果转换为excle
test = exe_sql('输入sql')
test.to_excel('./表名', index=False, encoding='gbk')
odps很重要的一个点是分区的概念,由于分区的存在会使得查询的时候数据量少很多,查询效率更高,但是有的时候分区的数据满足不了查询,需要全表查询的时候,在odpsSQL执行窗口中可以用这条语句来默认全表扫描,
SET odps.sql.allow.fullscan=true;
但是在Python中,如果这条语句直接放在SQL中,则执行的时候会报错,我们可以通过以下办法来解决
from odps import options
options.sql.settings = {"odps.sql.submit.mode": "script"}
设置一下模式,这样就可以了
在运行时如果需要设置参数,您可以通过设置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
如果您想要限制下载数据的规模,可以为open_reader增加limit选项, 或者设置 options.tunnel.limit_instance_tunnel = True 。如果未设置 options.tunnel.limit_instance_tunnel,MaxCompute会自动打开数据量限制,此时,可下载的数据条数受到Project配置的Tunnel下载数据规模数限制, 通常该限制为10000条。
# 直接使用 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()
您可以通过n_process指定使用进程数。
说明:目前多进程加速在 Windows 下无法使用。
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)