PyODPS 中使用 Python UDF

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: PyODPS 中使用 Python UDF 包含两方面,一个是直接使用,也就是在 MaxCompute SQL 中使用;一个是间接的方式,也就是 PyODPS DataFrame,这种方式你不需要直接写 Python UDF,而是写普通的 Python 函数或者类。

PyODPS 中使用 Python UDF 包含两方面,一个是直接使用,也就是在 MaxCompute SQL 中使用;一个是间接的方式,也就是 PyODPS DataFrame,这种方式你不需要直接写 Python UDF,而是写普通的 Python 函数或者类。下面我们分开说明。

作为准备工作,我们需要 ODPS 入口,可以通过直接初始化,或者使用 room 机制 加载。

from odps import ODPS

o = ODPS('your-access-id', 'your-access-key', 'your-project')

MaxCompute SQL 中使用 Python UDF

首先,我们需要写一个 Python 文件,假设我们就是把某一列按 csv 格式放的一列转成 json 格式。

import json

from odps.udf import annotate

@annotate('string->string')
class Transform(object):
    def evaluate(self, x):
        columns = list('abc')
        d = dict(zip(columns, x.split(',')))
        return json.dumps(d)

假设这个文件叫 my.py,接下来我们就需要创建 py 资源。

r = o.create_resource('csv_to_json.py', 'py', fileobj=open('my.py'))

fileobj 参数也可以是 str 类型,就是表示文件的内容

接着我们就可以创建 Python UDF 了。

o.create_function('csv_to_json', class_type='csv_to_json.Transform', resources=[r])

这里我们指定了函数名叫 csv_to_json,主类使我们上传的 csv_to_json.py 文件里的 Transform 类。

现在我们就可以在 MaxCompute SQL 中调用这个 UDF 了。

o.execute_sql('select csv_to_json(raw) from pyodps_test_udf')

这样我们就完成了在 PyODPS 中使用 MaxCompute SQL + Python UDF 的整个过程。

PyODPS DataFrame

对于 PyODPS DataFrame 来说,用户只需要写普通的 Python 函数或者类,在函数或者类里,甚至可以读取全局变量,这样给开发带来了极大的方便。

和上面的例子目标相同,我们定义一个 transform 函数即可。然后我们对于 DataFrame 的一列调用 map 方法来应用这个函数。

passed_columns = list('abc')  # 可以从数据库中读取或者写死

def transform(x):
    import json
    d = dict(zip(passed_columns, x.split(',')))
    return json.dumps(d)

df.raw.map(transform)
In [30]: df
     raw
0  1,2,3
1  4,5,6
2  7,8,9

In [31]: df.raw.map(transform)
                              raw
0  {"a": "1", "c": "3", "b": "2"}
1  {"a": "4", "c": "6", "b": "5"}
2  {"a": "7", "c": "9", "b": "8"}

实际上,PyODPS DataFrame 在用 MaxCompute 执行的时候,也会创建 Python UDF 来实现这个功能,但用户不需要去创建文件、资源和函数这些过程,一切都是 Python 原生函数和类,整个过程相当顺畅。

另外可以看到,在上面的 my.py 里,我们也是定义了一个 columns 参数的,而如果这个参数是通过变量传进去的话,在 Python UDF 里非常麻烦,可能常常需要用一些 tricky 的方法,比如写到某个文件资源,然后在 UDF 里读取之类的。而对于 DataFrame 来说,完全没有这个问题,我们可以自由读取全局变量。

不过要注意的是,这个全局变量是被序列化到各个机器上的,所以你修改它不会全局生效。

好了,还有什么问题可以随时和我们取得联系。

028df8ba14c11b8a5ea8c71eaad0d5e6fcd6dafc_jpeg

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
目录
相关文章
|
22天前
|
数据处理 Apache 数据库
将 Python UDF 部署到 Apache IoTDB 的详细步骤与注意事项
【10月更文挑战第21天】将 Python UDF 部署到 Apache IoTDB 中需要一系列的步骤和注意事项。通过仔细的准备、正确的部署和测试,你可以成功地将自定义的 Python UDF 应用到 Apache IoTDB 中,为数据处理和分析提供更灵活和强大的支持。在实际操作过程中,要根据具体情况进行调整和优化,以确保实现最佳的效果。还可以结合具体的代码示例和实际部署经验,进一步深入了解和掌握这一过程。
20 2
|
3月前
|
监控 Java Serverless
Serverless 应用的监控与调试问题之PyFlink对于Python UDF的性能如何提升
Serverless 应用的监控与调试问题之PyFlink对于Python UDF的性能如何提升
|
4月前
|
分布式计算 大数据 Java
如何使用Python的pyodps库来进行跨项目空间重命名表名?
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
50 12
|
3月前
|
机器学习/深度学习 分布式计算 DataWorks
DataWorks产品使用合集之创建的UDF在业务流程中没有生效,但单独执行脚本是成功的,该怎么办
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
4月前
|
消息中间件 分布式计算 DataWorks
DataWorks产品使用合集之如何使用Python和阿里云SDK读取OSS中的文件
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
4月前
|
资源调度 分布式计算 DataWorks
DataWorks产品使用合集之如何使用Python UDF(User-Defined Function)来引用第三方模块
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
4月前
|
SQL 分布式计算 大数据
MaxCompute产品使用合集之PyODPS Python类的开发如何用MC的资源
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
4月前
|
机器学习/深度学习 分布式计算 大数据
MaxCompute产品使用合集之是否可以将5个资源包统一写到同一个python UDF脚本
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
4月前
|
SQL 分布式计算 DataWorks
DataWorks操作报错合集之重新上传后只有SQL无法运行,而Python可以正常运行,是什么导致的
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
4月前
|
运维 DataWorks 安全
DataWorks产品使用合集之如何在本地环境中安装Python包
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。