您当前的位置:首页 > 计算机 > 编程开发 > Python

python操作odps

时间:09-01来源:作者:点击数:

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
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') #官方提供的接口
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()  # 阻塞直到完成。

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')
4、odps全表扫描在Python中怎么实现

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条。

设置读取结果为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()
设置读取速度(进程数)

您可以通过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)
方便获取更多学习、工作、生活信息请关注本站微信公众号城东书院 微信服务号城东书院 微信订阅号
推荐内容
相关内容
栏目更新
栏目热门
本栏推荐