# 基于Aura DataFrame的多模数据处理 #### 操作场景 多模数据引擎Aura是专为分析多模态类型数据而设计，支持数据的**边读边算**能力，避免了传统方式的频繁存盘操作，使得视频解码、向量化等预处理环节能够流水线式执行。提供自定义UDF功能，满足多模态场景定制化数据处理的需求。 本章节操作以表格数据处理为示例，介绍在AI DataLake使用多模数据引擎Aura和DataArts Notebook交互式地完成数据的处理与分析的操作。 在作业开发过程中使用到了DataArts Notebook镜像内置的Aura DataFrame SDK，可直接使用魔法命令创建连接。 #### 准备工作 - 已注册账号并实名认证，且账号不能处于欠费或冻结状态。 - 已开通AI DataLake服务并授权使用云服务资源。 - 已开通LakeFormation、OBS权限并进行了委托确认。 - 已开通DataArts Studio服务，并创建了DataArts Studio实例及工作空间。 - 已购买Notebook资源组并创建Notebook实例。 #### 操作流程 本节介绍Aura Job的基本开发流程。技术实现流程： 1. **环境准备** 1. 在AI DataLake管理控制台创建工作空间，用于提供独立的作业运行环境。 2. 在AI DataLake管理控制台创建计算资源池，为作业的运行提供计算资源。 3. 在AI DataLake管理控制台创建端点，配置Aura引擎与计算资源池的关联关系。 2. **绑定计算资源和数据开发** 1. 在Notebook页面或DataArts Studio管理中心绑定Aura计算资源。 2. 在Notebook页面为当前Notebook指定已绑定的Aura计算资源。 3. 在Notebook交互式开发页面使用**%aura_frame**魔法命令建立与多模态数据引擎Aura的连接。 3. **算子注册** 注册数据处理算子，构建数据处理流水线： 表格处理算子：对表格数据进行自定义处理。 4. **执行作业后查看结果** 1. 执行算子操作，处理数据。 2. 在Notebook交互式开发界面查看执行结果。 #### 步骤一：规划并创建OBS并行文件系统 AI DataLake Aura通过OBS服务实现数据存储，需要先在OBS控制台进行桶及文件夹创建，并导入样例数据。 1. 登录[华为云OBS管理控制台](https://console.huaweicloud.com/obs/)。 2. 选择"并行文件系统 \> 创建并行文件系统"，进入创建页面，配置相关参数后单击"立即创建"。 - 文件系统名称：根据界面要求设置并行文件系统名称，例如"aura-serverless"。 - 其他参数根据实际情况选择。 3. 在并行文件系统页面，单击已创建的文件系统名称，例如"aura-serverless"。 4. 在左侧导航栏选择"文件"，单击"新建文件夹"，填写待创建的文件夹名称，单击"确定"。继续单击该新创建的文件夹名称，单击"新建文件夹"，可以创建其子文件夹。 创建用于存放元数据的路径，例如： - Catalog存储路径：aura-serverless/catalog1 - 数据库存储路径：aura-serverless/catalog1/database1 #### 步骤二：规划并创建LakeFormation实例、Catalog、数据库 AI DataLake Aura通过LakeFormation服务管理数据源，需要在LakeFormation购买实例，并配置该实例的Catalog、数据库信息。 1. 登录[华为云LakeFormation管理控制台](https://console.huaweicloud.com/lake-formation/)。 2. 在"总览"页面右上角单击"购买实例"，配置相关参数购买LakeFormation实例。 3. 在页面左上角选择切换到该实例。 图1切换目标实例 ![](https://support.huaweicloud.com/qs-aidatalake/zh-cn_image_0000002593985657.png) 4. 创建Catalog。 1. 在左侧导航栏选择"数据目录"。 2. 在Catalog页面单击"创建"，配置以下参数后，单击"提交"。 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | Catalog名称 | lakeformation_for_auratest | 填写待创建Catalog名称。 只能包含字母、数字和下划线，长度为1\~256个字符。 | | Catalog类型 | DEFAULT | 选择Catalog类型： - DEFAULT：即默认数据目录，用于管理存储在OBS中的数据资产。 - CLICKHOUSE：CLICKHOUSE Catalog是用于连接外部ClickHouse数据库的外部目录类型。 | | 选择位置 | obs://lakeformation-test/catalog1 | Catalog数据存储在OBS桶中的位置。可选参数。 根据实际需要选择"并行文件系统"或"对象存储桶"，并选择位置后，单击"确定"。 - 如果配置该参数，则所选位置只能以"obs://"开头，且必须包含一个存储对象，例如选择"obs://lakeformation-test/catalog1"。如果没有合适的OBS桶，可以单击"前往OBS创建"进行创建。 - 该路径不能与其他LakeFormation实例元数据存储路径重复，以免造成数据冲突。 - 建议选择未被其他Catalog选中的路径。 | | 描述 | 按需配置 | 所创建Catalog的描述信息。 长度为0\~4000字节，1个中文字符对应3个字节。 | 3. 创建完成后，即可在"Catalog"页面查看相关信息。 5. 创建数据库。 1. 单击待创建数据库所属Catalog名称，在"数据库"页签查看当前Catalog中包含的数据库。 如果当前已包含名称为"default"的数据库，则跳过数据库的创建操作。 2. 在"数据库"页签单击"创建"，配置相关参数后，单击"提交"。 ![](https://support.huaweicloud.com/qs-aidatalake/public_sys-resources/caution_3.0-zh-cn.png) 在正式使用新Catalog前，请先手动创建default数据库并指定合法OBS路径，可避免后续系统自动创建带来的隐藏风险。 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | 库名称 | default | 填写待创建数据库名称。 只能包含中文、字母、数字、下划线、中划线，长度为1\~128个字符。 | | 所属Catalog | lakeformation_for_auratest | 待创建数据库所属Catalog。 本例选择步骤[4]创建的Catalog。 | | 选择位置 | obs://lakeformation-test/catalog1/default | 数据库信息存储在OBS桶中的位置。 根据实际需要选择"并行文件系统"或"对象存储桶"，并选择位置后，单击"确定"。 - 所选位置只能以"obs://"开头，且必须包含一个存储对象，例如选择"obs://lakeformation-test/catalog1/default"。如果没有合适的OBS桶，可以单击"前往OBS创建"进行创建。 - 该路径必须与所属Catalog的存储路径（即创建Catalog时配置的"选择位置"参数）不同。 - 该路径不能与其他LakeFormation实例元数据存储路径重复，以免造成数据冲突。 - 如果所属Catalog配置了"数据库存储位置"参数，则此处该参数必须选择为所属Catalog"选择位置"的子路径、或"数据库存储位置"的子路径。 | | 描述 | 按需配置 | 所创建数据库的描述信息。 长度为0\~4000字节，1个中文字符对应3个字节。 | 3. 创建完成后，即可在"数据库"页面查看详细信息。 #### 步骤三：创建工作空间 1. 登录[AI DataLake管理控制台](https://console.huaweicloud.com/fabric/)。 2. 在左侧导航栏中，单击工作空间区域。 3. 在下拉列表中，选择"创建工作空间" 。 4. 配置工作空间的相关参数： 表1创建工作空间参数说明 | 类型 | 参数 | 示例 | 说明 | |:---|:---|:---|:---| | 基础信息 | 工作空间名称 | workspace_auratest | 工作空间的具体名称。 - 名称只能包含字母、中文、数字、中划线、下划线。 - 输入长度为4\~32个字符。 工作空间名称不区分大小写，系统会自动转换为小写。 | | 基础信息 | 描述 | 智能驾驶数据分析 | 对工作空间的简要描述。 | | 基础信息 | 企业项目 | default | 如果所建工作空间属于企业项目，可选择对应的企业项目。企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理。 关于如何设置企业项目请参考《[企业管理用户指南](https://support.huaweicloud.com/usermanual-em/zh-cn_topic_0108763975.html)》。 说明： 只有开通了企业管理服务的用户才显示该参数。 | | 多模数据管理 | LakeFormation实例 | lakeformation_for_auratest | 选择当前工作空间关联的LakeFormation实例，即[步骤二：规划并创建LakeFormation实例、Catalog、数据库](https://support.huaweicloud.com/qs-aidatalake/aidatalake_01_0030.html#aidatalake_01_0030__section72971419177)创建的LakeFormation实例。 每个工作空间需关联1个LakeFormation实例，空间创建后已关联的实例不支持修改。 | 5. 确认所有配置信息无误。 单击"立即创建"，完成工作空间创建。 工作空间创建成功后，您可以在工作空间管理的列表中查看新创建的工作空间。 #### 步骤四：创建计算资源池 1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间为[步骤三：创建工作空间](https://support.huaweicloud.com/qs-aidatalake/aidatalake_01_0030.html#aidatalake_01_0030__section18692137202515)中所创建的空间。 2. 单击左侧导航栏的"资源管理"，选择"计算资源池"，单击右上方"购买"，进入购买计算资源池页面。 3. 在"购买计算资源池"界面，填写具体参数，参数填写参考[表2](https://support.huaweicloud.com/qs-aidatalake/aidatalake_01_0030.html#aidatalake_01_0030__table67098261452)。 表2购买计算资源池参数说明 | 参数 | 示例 | 说明 | |:---|:---|:---| | 计费模式 | 按需计费 | - **包年/包月：** 预付费模式，按订单的购买周期计费。在购买周期内资源独享，空闲（无作业运行）时不会释放，使用体验更佳，价格比按需计费模式更优惠。 适用于可预估资源使用周期的场景，例如已完成开发进入生产阶段的项目，推荐使用包年/包月计费模式。 - **按需计费**：即后付费模式，按实际使用量计费，在购买周期内资源独享，空闲时资源不被释放。 | | 资源池名称 | resource_pool_for_auratest | 计算资源池的具体名称。 - 名称只能包含数字、小写英文字母和中划线，且只能以字母开头、以字母或数字结尾。 - 输入长度不能超过63个字符。 | | CPU 资源 | 本例配置8个通用计算标准型实例。 | 勾选CPU资源后，选择资源规格并配置购买的实例数量。 CPU是通用型计算资源，适用于各种类型的计算任务。 - CPU资源的特点：通用性强，适合各种类型的计算任务。相比于GPU和NPU的成本更低。擅长处理顺序执行的任务。 - CPU资源的适用场景：ETL数据抽取、转换、加载；轻量计算场景，例如小规模数据处理、脚本运行，日志分析场景，例如日志采集、解析、统计分析。 - CPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html#section1) | | GPU 资源 | 本例不购买GPU实例 | 勾选GPU资源后，选择资源规格并配置购买的实例数量。 GPU是图形处理器，适用于图形渲染和大规模并行计算场景，适合深度学习训练和科学计算。GPU拥有成百上千个计算核心，可以同时处理大量简单计算任务。 - GPU资源的特点：支持并行计算，适用于批处理作业场景，数千个核心同时计算，处理效率高。天然适合深度学习、神经网络、AI计算场景。 - GPU资源的适用场景：深度学习，例如神经网络训练、模型调优场景；图形处理场景，例如图像识别、目标检测；视频处理场景，例如视频分析、转码、视频渲染场景，等其他AI科学计算场景。 - GPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html#section2) | | NPU 资源 | 本例不购买NPU资源 | 勾选NPU资源后，选择资源规格并配置购买的实例数量。 NPU适用于AI计算场景。NPU资源采用架构优化和指令集，专门加速AI推理任务，具有高能效比的特点。 - NPU资源的特点：AI计算场景专用，具备高性能、低延迟、推理成本更优的特点。 - NPU资源的适用场景：AI计算场景、图像识别场景，推荐系统等AI计算相关场景。 - NPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html#section3) | | AI DataLake 网络 | 自定义 | 选择AI DataLake资源池所属网络，该网络基于虚拟私有云（VPC）进行封装。如果不存在可使用的网络，也可单击"创建网络"进行创建。 **一个工作空间仅支持创建一个网络。** | 4. 参数填写完成后，单击"立即购买"，在界面上确认当前配置是否正确。 5. 单击"提交"完成创建，等待资源池状态变成"可用"表示当前资源池创建成功。 #### 步骤五：创建运行作业的Aura端点 1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间为[步骤三：创建工作空间](https://support.huaweicloud.com/qs-aidatalake/aidatalake_01_0030.html#aidatalake_01_0030__section18692137202515)新创建的空间。 2. 在左侧导航栏选择"引擎端点 \> 多模数据引擎Aura"进入Aura引擎端点列表页面。 3. 单击页面右上角的"创建端点"，配置以下参数并单击"立即创建"。 表3创建Aura引擎端点 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | 端点类型 | AuraJob | 选择端点类型。 AuraJob：用于执行定时调度任务，保障大规模数据稳定处理。 | | 端点版本 | v2.0 | 选择端点版本，基于Aura DataFrame的多模数据处理需选择"v2.0"。 - v1.0：自定义镜像算子，智算、通算混合资源池统一调度，支持Job级弹性。 - v2.0（推荐）：细粒度Actor级算子编排调度，数据处理Pipeline执行，性能和效率极大提升。 | | 端点名称 | aura-endpoint-test001 | 输入端点名称，是端点的唯一标识符，不可与已存在的端点重名，且创建后不支持修改。 - 名称只能包含小写字母、数字、中划线，且只能以字母开头，以字母或数字结尾。 - 输入长度不能超过63个字符。 | | 端点显示名称 | 智能驾驶团队专用Aura端点-02 | 输入端点显示名称，创建后可修改，用于用户界面展示，方便用户识别和记忆。 - 名称只能包含中文、英文、数字、中划线。 - 输入长度不能超过63个字符。 | | 资源使用模式 | 混合模式 | 选择资源使用模式。 - 预留资源：独享性能，成本最优，确保业务基线稳定。 适用于负载稳定业务场景。 - 混合模式：基线保障，自动扩容。优先消耗预留资源，高峰期自动触发弹性补位。 适用于有规律波动的业务。 | | 选择计算资源池 | resource_pool_for_auratest | 在下拉框中选择[步骤四：创建计算资源池](https://support.huaweicloud.com/qs-aidatalake/aidatalake_01_0030.html#aidatalake_01_0030__section1565751013319)已创建的资源池。 | | 工作组配置 | - | 配置Aura端点的工作组资源。 - 资源规格：在下拉框中选择需要配置的资源规格，支持选择CPU、GPU、NPU资源。 - 每Worker卡数：配置Worker卡数。 分配了GPU/NPU资源的每个Worker的GPU/NPU卡数。只有GPU和NPU场景可以配置卡数，CPU场景不支持配置。且GPU/NPU场景不支持调整Worker CPU和内存。 - 每Worker CPU及内存：配置Worker CPU和内存。 分配了CPU资源的工作组，不支持自定义GPU/NPU卡数。 - Worker 数量 (Min-Max)：每个工作组的Worker最大值和最小值。 - 预留资源：Worker最小值需与最大值相同，且需大于0。 - 混合模式：Worker最小值需小于等于最大值，且需大于0。 可单击"添加工作组"添加多个工作组配置，也可单击"删除"删除工作组，至少需保留一个工作组。 | | 选择OBS并行文件系统 | obs-aura | 选择用于存储作业运行过程中的结果缓存和日志文件的OBS并行文件系统，端点创建后不支持修改。 如果没有可用的OBS桶，可单击"新建OBS并行文件系统"进行创建。具体操作，请参见[创建并行文件系统](https://support.huaweicloud.com/usermanual-obs/obs_13_0002.html)。 | | 日志对接LTS | 勾选"日志对接LTS" | 是否对接LTS。对接LTS后日志会投递到云日志服务（Log Tank Service，简称LTS）进行管理。可以使用LTS对云服务日志进行关键词搜索、运营数据统计分析、运行状况监控告警等多种操作。 勾选该参数后，还需配置"日志组"和"日志流"参数。 | | 日志组 | container_aurajob_test | 在下拉框中选择日志组，如果下拉框中没有可选的日志组，可以单击"创建日志组"进行创建。 日志组（LogGroup）是云日志服务进行日志管理的基本单位，用于对日志流进行分类，一个日志组下面可以创建多个日志流。日志组本身不存储任何日志数据，仅方便管理日志流，每个账号下可以创建100个日志组。 | | 日志流 | container_job_log | 在下拉框中选择日志流，如果下拉框中没有可选的日志流，可以单击"创建日志流"进行创建。 云日志服务是以日志流（LogStream）作为日志管理维度。日志采集后，以日志流为单位，将不同类型的日志分类存储在不同的日志流上，方便对日志进一步分类管理。 | 4. 端点创建后，可在列表中查看相关信息，端点状态变为"运行中"后即可提交作业到该端点中运行。 #### 步骤六：连接多模态数据引擎Aura的端点 1. 在AI DataLake管理控制台页面，单击导航栏下方的"DataArts Studio"，进入DataArts Studio作业开发界面。 2. 在"数据开发 \> 作业开发"页面的右下角，单击"运行环境为空"，选择"新建环境"，配置以下参数并单击"创建"创建Notebook： - 运行环境名称：输入自定义名称。 - 所属资源组：选择已创建的Notebook资源组。 - 运行资源规格：选择所需的运行资源规格。 - 镜像选择：选择Notebook镜像。 3. 在"Notebook环境目录"区域右键单击"我的文件"选择"新建 Notebook"，配置Notebook名称并选择存放的目录，单击"确定"。 4. 在Notebook中的单元格执行以下魔法命令连接运行作业的Aura端点： ``` conn = %aura_frame --lf_catalog_name xxx --obs_bucket_name xxx --obs_directory_base xxx/test --lf_instance_id xxx --default_database xxx ``` 其中： - lf_catalog_name：连接的LakeFormation中的Catalog名称。 - --obs_bucket_name：用于存储UDF的OBS并行文件系统名称。 - --obs_directory_base：用于存储UDF的OBS路径。 - --lf_instance_id：连接的LakeFormation中的实例ID。 - --default_database：连接的LakeFormation中的默认数据库。 单击左上角的运行按钮运行命令连接Aura端点，回显以下信息，即表示连接Aura端点成功： ``` [INFO] [Aura Frame] 2026-04-29 09:21:18 Setting guc options: 'SET timezone = UTC; SET obs_result_format = 2;SET current_schema = xxx;' [INFO] [Aura Frame] 2026-04-29 09:21:19 Created session 019dd6d3-7fe1-743d-8017-ce7974125079 successfully. ``` #### 步骤七：导入Python和Aura DataFrame SDK依赖包 在Notebook中的新增单元格，执行以下命令导入Python和Aura DataFrame SDK依赖包： ``` import sys import requests import pandas as pd import aura_frame as aura from aura_frame.multimodal.types import image, embedding from aura_frame.ibis.backends.sql.datatype import advance_dtype ``` 单击左上角的运行按钮运行命令。 #### 步骤八：创建数据表 1. 在Notebook中的新增单元格，执行以下命令创建数据表： ``` df = pd.DataFrame({ 'a': [1], 'b': [3], 'c': [1] }) ds = conn.from_pandas(df) ds.show() ``` 2. 单击左上角的运行按钮运行命令，命令运行后，结果示例如下则表示创建表成功： ``` generated sql statement: CREATE TEMP TABLE aura_pandas_dataset_vp2r6lk7fna7zaldkcfsk3pajm ( "a" BIGINT,"b" BIGINT,"c" BIGINT ) STORE AS PANDAS [INFO] [Aura Frame] 2026-04-29 09:26:52 Loading dataset pg_temp.aura_pandas_dataset_vp2r6lk7fna7zaldkcfsk3pajm [INFO] [Aura Frame] 2026-04-29 09:26:55 read pandas aura_pandas_dataset_vp2r6lk7fna7zaldkcfsk3pajm [INFO] [Aura Frame] 2026-04-29 09:26:55 SQL statement for executing: SELECT * FROM "pg_temp"."aura_pandas_dataset_vp2r6lk7fna7zaldkcfsk3pajm" AS "t0" LIMIT 10 [INFO] [Aura Frame] 2026-04-29 09:26:58 Running SQL statement, id is 019dd6d8-a5a2-747c-901b-1fb08d25f379 a b c 0 1 3 1 ``` #### 步骤九：定义和注册UDF 1. 在Notebook中的新增单元格，执行以下命令定义和注册UDF： ``` target_database = xxx import ibis.expr.datatypes as dt import aura_frame as aura func_signature = aura.Signature( parameters=[ aura.Parameter(name="start", annotation=int), aura.Parameter(name="endnum", annotation=int), aura.Parameter(name="step", annotation=int) ], return_annotation=dt.Struct({"col1": int}), ) @aura.udf.python(database=target_database, signature=func_signature) def py_generate_series(start, endnum, step): current = start while current <= endnum: yield {'col1': current} current += step # 如果未注册UDF，则无需执行删除操作，需注释掉delete_function操作 conn.delete_function("py_generate_series", database=target_database) fn = conn.create_table_function( py_generate_series, name="py_generate_series", database=target_database, signature=func_signature ) ``` "target_database"表示LakeFormation中用于管理注册的UDF的信息的数据库名称。 2. 单击左上角的运行按钮运行命令，命令运行后，结果示例如下则表示定义和注册UDF成功： ``` [INFO] [Aura Frame] 2026-04-29 09:26:58 Start to register function xxx.py_generate_series [INFO] [Aura Frame] 2026-04-29 09:26:58 Function main entry code has been serialized by cloudpickle. [INFO] [Aura Frame] 2026-04-29 09:26:58 The Python package dependencies have been resolved. [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types(4.0 KB) to archive zipfile as aura_frame/multimodal/types [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/__init__.py(0 bytes) to archive zipfile as aura_frame/multimodal/types/__init__.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/audio.py(8.93 KB) to archive zipfile as aura_frame/multimodal/types/audio.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/embedding.py(798 bytes) to archive zipfile as aura_frame/multimodal/types/embedding.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/image.py(5.83 KB) to archive zipfile as aura_frame/multimodal/types/image.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/vectorized.py(508 bytes) to archive zipfile as aura_frame/multimodal/types/vectorized.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types/video.py(10.83 KB) to archive zipfile as aura_frame/multimodal/types/video.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Appending /opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/utils.py(1.84 KB) to archive zipfile as aura_frame/multimodal/utils.py [INFO] [Aura Frame] 2026-04-29 09:26:58 Function py_generate_series's archive zipfile has been uploaded to OBS. [INFO] [Aura Frame] 2026-04-29 09:26:58 SQL statement for registering function is CREATE FUNCTION xxx.py_generate_series(start BIGINT, endnum BIGINT, step BIGINT) RETURNS SETOF STRUCT<"col1":BIGINT> LANGUAGE PYTHON RUNTIME_VERSION = '3.11' STRICT VOLATILE IMPORTS = ('https://xxx.com/xxx/xxx/py_generate_series/py_generate_series_20260429092658.zip') HANDLER = 'py_generate_series' AS $$ # Comment for UDF 'py_generate_series' # === Function Metadata === # UDTFMeta( # _runtime_version='3.11', # _runtime_env=RuntimeEnv({}), # _registered=False, # _input_type=, # _replace=False, # _temporary=False, # _if_not_exists=False, # _strict=True, # _volatility=, # _imports=('/opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/types', '/opt/cloud/3rdComponent/python3.11/lib/python3.11/site-packages/aura_frame/multimodal/utils.py'), # _packages=(), # _register_type=None, # _comment=None, # _work_group_name=None, # _database='xxx', # _func_name='py_generate_series', # _signature= struct>, # _udf=, # _module='__main__', # ) # === Function Source Code === # @aura.udf.python(database=target_database, signature=func_signature) # def py_generate_series(start, endnum, step): # current = start # while current <= endnum: # yield {'col1': current} # current += step import cloudpickle py_generate_series = cloudpickle.loads(bytes.fromhex("xxx")) $$ COMMENT '{"description": null, "pretty_name": "py_generate_series"}' [INFO] [Aura Frame] 2026-04-29 09:27:01 User-defined function xxx.py_generate_series registered successfully. ``` #### 步骤十：使用UDF向数据表中写入数据 1. 在Notebook中的新增单元格，执行以下命令使用UDF向数据表中写入数据： ``` ds = ds.flat_map( fn="py_generate_series", on=[ds.a, ds.b, ds.c], as_col='new_col' ) ds.show() ``` 2. 单击左上角的运行按钮运行命令，命令运行后，结果示例如下则表示向数据表中写入数据成功： ``` [INFO] [Aura Frame] 2026-04-29 09:27:01 SQL statement for executing: SELECT "t2"."a", "t2"."b", "t2"."c", "t2"."new_col" FROM (SELECT "t1"."a", "t1"."b", "t1"."c", "t1"."udtf_tmp_col", ("t1"."udtf_tmp_col")."col1" AS "new_col" FROM (SELECT "t0"."a", "t0"."b", "t0"."c", xxx.PY_GENERATE_SERIES("t0"."a", "t0"."b", "t0"."c") WITH ARGUMENTS(cpu = 0.5, memory = 1024, arch = 'x86') AS "udtf_tmp_col" FROM "pg_temp"."aura_pandas_dataset_vp2r6lk7fna7zaldkcfsk3pajm" AS "t0") AS "t1") AS "t2" LIMIT 10 [INFO] [Aura Frame] 2026-04-29 09:27:05 Running SQL statement, id is 019dd6d8-be56-71e9-9196-d7f7e04a4c8f a b c new_col 0 1 3 1 1 1 1 3 1 2 2 1 3 1 3 ``` #### 步骤十一：关闭数据库连接 在使用数据库连接完成相应的数据操作后，需要关闭数据库连接。 ``` conn.close() ```