[1172]python操作odps

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

1、安装ODPS

pip install ODPS
1

2、连接阿里云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') #官方提供的接口
1
2
3
4
5
6
7

3、执行sql

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()  # 阻塞直到完成。
1
2
3
4
5
6
7

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') #获取表数据实例
1
2
3
4
5

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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.4、将sql结果转换为excle

test = exe_sql('输入sql')
test.to_excel('./表名', index=False, encoding='gbk')
1
2

4、odps全表扫描在Python中怎么实现

odps很重要的一个点是分区的概念，由于分区的存在会使得查询的时候数据量少很多，查询效率更高，但是有的时候分区的数据满足不了查询，需要全表查询的时候，在odpsSQL执行窗口中可以用这条语句来默认全表扫描，

SET odps.sql.allow.fullscan=true;
1

但是在Python中，如果这条语句直接放在SQL中，则执行的时候会报错，我们可以通过以下办法来解决

from odps import options

options.sql.settings = {"odps.sql.submit.mode": "script"}
1
2
3

设置一下模式，这样就可以了

设置运行参数

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

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

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

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

设置读取数据规模

如果您想要限制下载数据的规模，可以为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()
1
2
3
4

设置读取速度（进程数）

您可以通过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)
1
2
3
4
5

参考：https://help.aliyun.com/document_detail/34615.html
https://blog.csdn.net/T_qiuqiu/article/details/126348157
https://www.jianshu.com/p/cc7d0461f6d5
https://blog.csdn.net/weixin_38938108/article/details/125853375
https://blog.csdn.net/weixin_42261073/article/details/103199562

实例操作：https://help.aliyun.com/document_detail/27830.html
MaxCompute客户端常用命令：https://blog.csdn.net/yitian_z/article/details/105540527
pyodps：https://pyodps.readthedocs.io/en/latest/base-instances.html?highlight=instance
使用Logview 2.0查看Job运行信息：https://help.aliyun.com/document_detail/183576.html