2025年4月2日 星期三 乙巳(蛇)年 正月初三 设为首页 加入收藏
rss
您当前的位置:首页 > 计算机 > 编程开发 > Python

python操作odps

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

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