文档首页/ 智能数据湖 AI DataLake/ 快速入门/ 基于Aura DataFrame的多模数据处理
更新时间：2026-10-16 GMT+08:00
查看PDF
分享

基于Aura DataFrame的多模数据处理

操作场景

多模数据引擎Aura是专为分析多模态类型数据而设计，支持数据的边读边算能力，避免了传统方式的频繁存盘操作，使得视频解码、向量化等预处理环节能够流水线式执行。提供自定义UDF功能，满足多模态场景定制化数据处理的需求。

本章节操作以表格数据处理为示例，介绍在AI DataLake使用AuraJobV2和DataArts Studio Notebook交互式地完成数据的处理与分析的操作。

在作业开发过程中使用到了DataArts Studio Notebook镜像内置的Aura DataFrame SDK，可直接使用魔法命令创建连接。

DataArts Studio Notebook当前处于受限使用阶段，如需使用请提交工单申请开通。

准备工作

  • 已注册账号并实名认证，且账号不能处于欠费或冻结状态。
  • 已开通AI DataLake服务并授权使用云服务资源。
  • 已开通LakeFormation、OBS权限并进行了委托确认。
  • 已开通DataArts Studio服务，并创建了DataArts Studio实例及工作空间。
  • 已购买Notebook资源组并创建Notebook实例。

操作流程

本节介绍AuraJobV2的基本开发流程。技术实现流程：

  1. 环境准备
    1. 在AI DataLake管理控制台创建工作空间，用于提供独立的作业运行环境。
    2. 在AI DataLake管理控制台创建计算资源池，为作业的运行提供计算资源。
    3. 在AI DataLake管理控制台创建端点，配置AuraJobV2与计算资源池的关联关系。
  2. 绑定计算资源和数据开发
    1. 在Notebook页面或DataArts Studio管理中心绑定Aura计算资源。
    2. 在Notebook页面为当前Notebook指定已绑定的Aura计算资源。
    3. 在Notebook交互式开发页面使用%aura_frame魔法命令建立与AuraJobV2端点的连接。
  3. 算子注册

    注册数据处理算子，构建数据处理流水线：

    表格处理算子：对表格数据进行自定义处理。

  4. 执行作业后查看结果
    1. 执行算子操作，处理数据。
    2. 在Notebook交互式开发界面查看执行结果。

步骤一：规划并创建OBS并行文件系统

AI DataLake AuraJobV2通过OBS服务实现数据存储，需要先在OBS控制台进行桶及文件夹创建，并导入样例数据。

  1. 登录华为云OBS管理控制台。
  2. 选择“并行文件系统 > 创建并行文件系统”，进入创建页面，配置相关参数后单击“立即创建”。
    • 文件系统名称：根据界面要求设置并行文件系统名称，例如“aura-serverless”。
    • 其他参数根据实际情况选择。
  3. 在并行文件系统页面，单击已创建的并行文件系统名称，例如“aura-serverless”。
  4. 在左侧导航栏选择“文件”，单击“新建文件夹”，输入文件夹名称，单击“确定”。
    支持直接输入多级路径（用“/”分隔各层级文件夹名称）一次性创建多层级文件夹。例如：
    • Catalog存储路径：aura-serverless/catalog1
    • 数据库存储路径：aura-serverless/catalog1/database1

步骤二：规划并创建LakeFormation实例、Catalog、数据库

AI DataLake AuraJobV2通过LakeFormation服务管理数据源，需要在LakeFormation购买实例，并配置该实例的Catalog、数据库信息。

  1. 登录华为云LakeFormation管理控制台。
  2. 在“总览”页面右上角单击“购买实例”，配置相关参数购买LakeFormation实例。具体操作，请参见创建LakeFormation实例。
  3. 在页面左上角选择切换到该实例。
    图1 切换目标实例
  4. 创建Catalog。
    1. 在左侧导航栏选择“数据目录”。
    2. 在Catalog页面单击“创建”，配置以下参数后，单击“提交”。

      参数

      示例

      参数说明

      Catalog名称

      lakeformation_for_auratest

      填写待创建Catalog名称。

      只能包含字母、数字和下划线，长度为1~256个字符。

      Catalog类型

      DEFAULT

      选择Catalog类型：

      • DEFAULT：即默认数据目录，用于管理存储在OBS中的数据资产。
      • CLICKHOUSE：CLICKHOUSE Catalog是用于连接外部ClickHouse数据库的外部目录类型。

      选择位置

      obs://aura-serverless/catalog1

      Catalog数据存储在OBS桶中的位置。可选参数。

      根据实际需要选择“并行文件系统”或“对象存储桶”，并选择位置后，单击“确定”。

      • 如果配置该参数，则所选位置只能以“obs://”开头，且必须包含一个存储对象，例如选择“obs://aura-serverless/catalog1”。如果没有合适的OBS桶，可以单击“前往OBS创建”进行创建。
      • 该路径不能与其他LakeFormation的Catalog数据存储目录重复，以免造成数据冲突。
      • 建议选择未被其他Catalog选中的路径。

      描述

      按需配置

      所创建Catalog的描述信息。

      长度为0~4000字节，1个中文字符对应3个字节。
    3. 创建完成后，即可在“Catalog”页面查看相关信息。
  5. 创建数据库。
    1. 单击待创建数据库所属Catalog名称，在“数据库”页签查看当前Catalog中包含的数据库。

      如果当前已包含名称为“default”的数据库，则跳过数据库的创建操作。

    2. 在“数据库”页签单击“创建”，配置相关参数后，单击“提交”。

      在正式使用新Catalog前，请先手动创建default数据库并指定合法OBS路径，可避免后续系统自动创建带来的隐藏风险。

      参数

      示例

      参数说明

      库名称

      database1

      填写待创建数据库名称。

      只能包含中文、字母、数字、下划线、中划线，长度为1~128个字符。

      所属Catalog

      lakeformation_for_auratest

      待创建数据库所属Catalog。

      本例选择4创建的Catalog。

      选择位置

      obs://aura-serverless/catalog1/database1

      数据库信息存储在OBS并行文件系统中的位置。

      根据实际需要选择“并行文件系统”，并选择位置后，单击“确定”。

      • 所选位置只能以“obs://”开头，且必须包含一个存储对象，例如选择“obs://aura-serverless/catalog1/database1”。如果没有合适的OBS桶，可以单击“前往OBS创建”进行创建。
      • 该路径必须与所属Catalog的存储路径（即创建Catalog时配置的“选择位置”参数）不同。
      • 该路径不能与其他LakeFormation实例的Catalog存储路径重复，以免造成数据冲突。
      • 如果所属Catalog配置了“数据库存储位置”参数，则此处该参数必须选择为所属Catalog“选择位置”的子路径、或“数据库存储位置”的子路径。

      描述

      按需配置

      所创建数据库的描述信息。

      长度为0~4000字节，1个中文字符对应3个字节。
    3. 创建完成后，即可在“数据库”页面查看详细信息。

步骤三：创建工作空间

  1. 登录AI DataLake管理控制台。
  2. 在左侧导航栏中，单击工作空间区域。
  3. 在下拉列表中，选择“创建工作空间” 。
  4. 配置工作空间的相关参数：
    表1 创建工作空间参数说明

    类型

    参数

    示例

    说明

    基础信息

    工作空间名称

    workspace_auratest

    工作空间的具体名称。

    • 名称只能包含字母、中文、数字、中划线、下划线。
    • 输入长度为4~32个字符。

    工作空间名称不区分大小写，系统会自动转换为小写。

    描述

    智能驾驶数据分析

    对工作空间的简要描述。

    企业项目

    default

    如果所建工作空间属于企业项目，可选择对应的企业项目。企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理。

    关于如何设置企业项目请参考《企业管理用户指南》。

    说明：

    只有开通了企业管理服务的用户才显示该参数。

    多模数据管理

    LakeFormation实例

    lakeformation_for_auratest

    选择当前工作空间关联的LakeFormation实例，即步骤二：规划并创建LakeFormation实例、Catalog、数据库创建的LakeFormation实例。

    每个工作空间需关联1个LakeFormation实例，空间创建后已关联的实例不支持修改。
  5. 确认所有配置信息无误。

    单击“立即创建”，完成工作空间创建。

    工作空间创建成功后，您可以在工作空间管理的列表中查看新创建的工作空间。

步骤四：创建计算资源池

  1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间为步骤三：创建工作空间中所创建的空间。
  2. 单击左侧导航栏的“资源管理 > 计算资源池”，单击右上方“购买”，进入购买计算资源池页面。
  3. 在“购买计算资源池”界面，配置参数。
    具体参数说明，请参见表2。
    表2 购买计算资源池参数说明

    参数

    示例

    说明

    计费模式

    按需计费
    • 包年/包月：预付费模式，按订单的购买周期计费。在购买周期内资源独享，空闲（无作业运行）时不会释放，使用体验更佳，价格比按需计费模式更优惠。

      适用于可预估资源使用周期的场景，例如已完成开发进入生产阶段的项目，推荐使用包年/包月计费模式。

    • 按需计费：按照实际运行时长计费（秒级计费，按小时结算）。资源归用户独占，可随时启用或释放。

    资源池名称

    resource_pool_for_auratest

    计算资源池的具体名称。

    • 名称只能包含数字、小写英文字母和中划线，且只能以字母开头、以字母或数字结尾。
    • 输入长度不能超过63个字符。

    CPU 资源

    本例配置8个通用计算标准型实例。

    勾选CPU资源后，选择资源规格并配置购买的实例数量。

    CPU是通用型计算资源，适用于各种类型的计算任务。

    • CPU资源的特点：通用性强，适合各种类型的计算任务。相比于GPU和NPU的成本更低。擅长处理顺序执行的任务。
    • CPU资源的适用场景：ETL数据抽取、转换、加载；轻量计算场景，例如小规模数据处理、脚本运行，日志分析场景，例如日志采集、解析、统计分析。
    • CPU资源的具体规格请参考产品规格

    GPU 资源

    本例不购买GPU实例

    勾选GPU资源后，选择资源规格并配置购买的实例数量。

    GPU是图形处理器，适用于图形渲染和大规模并行计算场景，适合深度学习训练和科学计算。GPU拥有成百上千个计算核心，可以同时处理大量简单计算任务。

    • GPU资源的特点：支持并行计算，适用于批处理作业场景，数千个核心同时计算，处理效率高。天然适合深度学习、神经网络、AI计算场景。
    • GPU资源的适用场景：深度学习，例如神经网络训练、模型调优场景；图形处理场景，例如图像识别、目标检测；视频处理场景，例如视频分析、转码、视频渲染场景，等其他AI科学计算场景。
    • GPU资源的具体规格请参考产品规格

    NPU 资源

    本例不购买NPU资源

    勾选NPU资源后，选择资源规格并配置购买的实例数量。

    NPU适用于AI计算场景。NPU资源采用架构优化和指令集，专门加速AI推理任务，具有高能效比的特点。

    • NPU资源的特点：AI计算场景专用，具备高性能、低延迟、推理成本更优的特点。
    • NPU资源的适用场景：AI计算场景、图像识别场景，推荐系统等AI计算相关场景。
    • NPU资源的具体规格请参考产品规格

    AI DataLake 网络

    自定义

    选择AI DataLake资源池所属网络，该网络基于虚拟私有云（VPC）进行封装。如果不存在可使用的网络，也可单击“创建网络”进行创建。

    一个工作空间仅支持创建一个网络。
  4. 参数填写完成后，单击“立即购买”，在界面上确认当前配置是否正确。
  5. 单击“提交”完成创建，等待资源池状态变成“可用”表示当前资源池创建成功。

步骤五：创建运行作业的AuraJobV2端点

  1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间。
  2. 在左侧导航栏选择“引擎端点 > 多模数据引擎 Aura”，进入Aura引擎端点列表页面。
  3. 单击页面右上角的“创建端点”，配置参数。
    关键参数说明如表3所示，其他参数可保持默认即可。
    表3 参数说明

    参数

    示例

    参数说明

    端点类型

    AuraJob

    默认选择“AuraJob”，不可修改。

    端点版本

    v2.0

    选择端点版本，基于Aura DataFrame的多模数据处理需选择“v2.0”。

    端点名称

    aura-endpoint-test001

    输入端点名称，是端点的唯一标识符，不可与已存在的端点重名，且创建后不支持修改。

    • 名称只能包含小写字母、数字、中划线，且只能以字母开头，以字母或数字结尾。
    • 输入长度不能超过63个字符。

    端点显示名称

    智能驾驶团队专用Aura端点-02

    输入端点显示名称，创建后可修改，用于用户界面展示，方便用户识别和记忆。

    • 名称只能包含中文、英文、数字、中划线。
    • 输入长度不能超过63个字符。

    资源使用模式

    混合模式

    选择资源使用模式。

    • 预留资源：独享性能，成本最优，确保业务基线稳定。

      适用于负载稳定业务场景。

    • 混合模式：基线保障，自动扩容。优先消耗预留资源，高峰期自动触发弹性补位。

      适用于有规律波动的业务。

    选择计算资源池

    resource_pool_for_auratest

    在下拉框中选择步骤四：创建计算资源池已创建的资源池。

    工作组配置

    -

    配置AuraJobV2端点的工作组资源。

    • 资源规格：本示例在下拉框中选择需要配置的CPU资源规格。
    • 每Worker资源配置：在下拉框中选择每个Worker的CPU和内存。
    • Worker数量 (Min-Max)：每个工作组的Worker最大值和最小值。

      Worker最小值需小于等于最大值，且需大于0。

    • 镜像：本示例选择预置镜像。将鼠标悬浮在“镜像”列，单击，在“预置镜像”页签中选择镜像名称，并勾选镜像版本，单击“确定”。

    选择结果集存储

    obs-aura

    选择用于存储作业运行过程中的结果缓存和日志文件的OBS并行文件系统，端点创建后不支持修改。

    如果没有可用的OBS并行文件系统，可单击“新建OBS并行文件系统”进行创建。具体操作，请参见创建并行文件系统。

    说明：

    结果集存储用于保存作业运行过程中的结果文件，您可以根据结果文件所需的保存时长手动定期清理或者在OBS并行文件系统中通过配置生命周期规则自定义存储时长。具体操作，请参见配置生命周期规则。

    选择临时存储资源

    storage-t1sa

    选择用于缓存Aura引擎的运行时程序的存储资源，端点创建后不支持修改。

    当计算任务需要访问或生成大量数据，且这些数据无法完全由计算节点本地存储或共享存储系统承载时，就需要增加计算资源组外的存储资源。

    Aura端点支持用户配置外部的存储资源，用于存储Aura端点运行作业的数据。

    在下拉列表中选择存储资源，如果没有可用的存储资源，请单击“购买存储资源”创建。具体操作，请参见购买存储资源。
  4. 单击“立即创建”，当端点状态变为“运行中”后，表示端点创建成功。

步骤六：连接AuraJobV2端点

  1. 在AI DataLake管理控制台页面，单击导航栏左下方的“DataArts Studio”，进入DataArts Studio作业开发界面。
  2. 新建运行环境。
    1. 在“数据开发 > 作业开发”页面的右下角，单击“运行环境为空”，选择“新建环境”，配置以下参数并单击“创建”创建Notebook。
      • 运行环境名称：输入自定义名称。
      • 所属资源组：选择已创建的Notebook资源组。
      • 运行资源规格：选择所需的运行资源规格。
      • 镜像选择：选择Notebook镜像。
    2. 在页面右上角，将鼠标悬浮在“未连接”上，在弹出提示框中，单击“选择运行环境”选择已创建的Notebook。
  3. 在“Notebook环境目录”区域右键单击“我的文件”选择“新建 Notebook”，配置Notebook名称并选择存放的目录，单击“确定”。
  4. 绑定计算资源。
    1. 在页面右上角，单击“Python”，在弹出的提示框中，单击“绑定计算资源”，配置参数后，单击“确定”。
      表4 绑定AuraJobV2端点参数

      参数

      示例

      说明

      资源类型

      Fabric SQL

      选择Notebook作业开发依赖的计算引擎。本示例选择“Fabric SQL”。

      资源名称

      aura_sql_966da08a

      输入计算资源名称。

      计算资源工作空间

      workspace_auratest

      选择的AuraJobV2端点所属的工作空间。

      端点

      aura-endpoint-test001

      选择AuraJobV2端点名称。
    2. 在页面右上角，单击“Python”，在弹出的提示框中，单击创建的计算资源名称。
  5. 在Notebook中的单元格输入以下魔法命令，连接运行作业的AuraJobV2端点。 
    con = %aura_frame --lf_catalog_name xxx --obs_bucket_name xxx --obs_directory_base xxx/test --default_database xxx
    表5 参数说明

    参数

    说明

    lf_catalog_name

    连接的LakeFormation中的Catalog名称。

    obs_bucket_name

    用于存储UDF的OBS并行文件系统名称。

    obs_directory_base

    用于存储UDF的OBS并行文件系统路径。

    default_database

    连接的LakeFormation中的默认数据库。

    单击左上角的运行命令，回显以下信息，即表示连接AuraJobV2端点成功：

    [INFO] [Aura Frame] 2026-09-30 11:21:36 Setting guc options: 'SET timezone = UTC; SET obs_result_format = 2;SET current_schema = database****;'
[INFO] [Aura Frame] 2026-09-30 11:21:36 Created session 01a0f054-f093-7059-8f71-bf173d08**** successfully.

步骤七：导入Python和Aura DataFrame SDK依赖包

  1. 在页面左上角，单击“新建单元格”。
  2. 在Notebook中的单元格，输入以下命令并单击左上角的，运行命令。 
    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. 在页面左上角，单击“新建单元格”。
  2. 在Notebook中的单元格，执行以下命令创建数据表： 
    import ibis

target_database = "xxx"   
table_name = "t1"
table_schema = ibis.schema({"a": "int32", "b": "int32", "c": "int32"})
con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
con.create_table(
    table_name,
    schema=table_schema,
    database=target_database,
    table_format="orc",
    if_not_exists=True,
)
con.sql(f'''
    INSERT INTO "{target_database}"."{table_name}" (a, b, c) VALUES
        (1,3,1)
    ''')
ds = con.load_dataset("t1", database=target_database)
  3. 单击左上角的，运行命令。
    命令运行后，结果示例如下则表示创建表成功。 
    [INFO] [Aura Frame] 2026-10-16 11:56:49 Creating table database1.t1
[INFO] [Aura Frame] 2026-10-16 11:56:49 CREATE TABLE IF NOT EXISTS "database1"."t1" ("a" INT, "b" INT, "c" INT) STORE AS orc
[INFO] [Aura Frame] 2026-10-16 11:56:58 Loading dataset database1.t1

步骤九：定义和注册UDF

  1. 在页面左上角，单击“新建单元格”。
  2. 在Notebook中的单元格，输入以下命令定义和注册UDF。 
    target_database = "xxx"    #替换为LakeFormation中用于管理已注册UDF的数据库名称
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}),
)
def py_generate_series(start, endnum, step):
    current = start
    while current <= endnum:
        yield {'col1': current}
        current += step
# 如果未注册UDF，则无需执行删除操作
con.delete_function("py_generate_series", database=target_database, if_it_exists=True)
con.create_table_function(
    py_generate_series,
    database=target_database,
    signature=func_signature
)
  3. 单击左上角的，运行命令。
    命令运行后，结果示例如下则表示定义和注册UDF成功。 
    [INFO] [Aura Frame] 2026-09-30 11:28:38 Deleting user-defined function database1.py_generate_series if exists
[INFO] [Aura Frame] 2026-09-30 11:28:42 Start to register function database1.py_generate_series
[INFO] [Aura Frame] 2026-09-30 11:28:42 Function main entry code has been serialized by cloudpickle.
[INFO] [Aura Frame] 2026-09-30 11:28:42 The Python package dependencies have been resolved.
[INFO] [Aura Frame] 2026-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/types(4.0 KB) to archive zipfile as aura_frame/multimodal/types
[INFO] [Aura Frame] 2026-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/types/__init__.py(0 bytes) to archive zipfile as aura_frame/multimodal/types/__init__.py
[INFO] [Aura Frame] 2026-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/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-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/types/embedding.py(798 bytes) to archive zipfile as aura_frame/multimodal/types/embedding.py
[INFO] [Aura Frame] 2026-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/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-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/types/vectorized.py(508 bytes) to archive zipfile as aura_frame/multimodal/types/vectorized.py
[INFO] [Aura Frame] 2026-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/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-09-30 11:28:42 Appending /opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/utils.py(1.84 KB) to archive zipfile as aura_frame/multimodal/utils.py
[INFO] [Aura Frame] 2026-09-30 11:28:42 Function py_generate_series's archive zipfile has been uploaded to OBS.
[INFO] [Aura Frame] 2026-09-30 11:28:42 SQL statement for registering function is 
CREATE    FUNCTION 
database1.py_generate_series(start BIGINT, endnum BIGINT, step BIGINT)
RETURNS SETOF STRUCT<"col1":BIGINT>
LANGUAGE PYTHON
RUNTIME_VERSION = '3.10'

STRICT
VOLATILE
IMPORTS = ('https://***.obs.cn-north-7.ulanqab.huawei.com/aura-serverless/catalog1/py_generate_series/py_generate_series_20260930112842.zip')


HANDLER = 'py_generate_series'
AS $$
# Comment for UDF 'py_generate_series'

# === Function Metadata ===
# UDTFMeta(
#     _runtime_version='3.10',
#     _runtime_env=RuntimeEnv({}),
#     _registered=False,
#     _input_type=<InputType.PYTHON: 4>,
#     _replace=False,
#     _temporary=False,
#     _if_not_exists=False,
#     _strict=True,
#     _volatility=<VolatilityType.VOLATILE: 3>,
#     _imports=('/opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/types', '/opt/cloud/3rdComponent/python/lib/python3.10/site-packages/aura_frame/multimodal/utils.py'),
#     _packages=(),
#     _register_type=None,
#     _comment=None,
#     _work_group_name=None,
#     _database='database1',
#     _func_name='py_generate_series',
#     _signature=<Signature (start: int, endnum: int, step: int) -> struct<col1: int64>>,
#     _udf=<function py_generate_series at 0x7fa718614e50>,
#     _module='__main__',
# )

# === Function Source Code ===
# 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("8005952e020000000000008c17636c6f75647069636b6c652e636c6f75647069636b6c65948c0e5f6d616b655f66756e6374696f6e9493942868008c0d5f6275696c74696e5f747970659493948c08436f6465547970659485945294284b034b004b004b044b024b63433081007c007d037c037c016b01721664017c036901560001007c037c0237007d037c037c016b0173076400530064005300944e8c04636f6c3194869429288c057374617274948c06656e646e756d948c0473746570948c0763757272656e749474948c212f746d702f6970796b65726e656c5f313735392f343139353131343239382e7079948c1270795f67656e65726174655f736572696573944b0c430c0280040108010a01080110fe942929749452947d94288c0b5f5f7061636b6167655f5f944e8c085f5f6e616d655f5f948c085f5f6d61696e5f5f94754e4e4e7494529468008c125f66756e6374696f6e5f7365747374617465949394681a7d947d942868178c1270795f67656e65726174655f736572696573948c0c5f5f7175616c6e616d655f5f948c1270795f67656e65726174655f736572696573948c0f5f5f616e6e6f746174696f6e735f5f947d948c0e5f5f6b7764656661756c74735f5f944e8c0c5f5f64656661756c74735f5f944e8c0a5f5f6d6f64756c655f5f9468188c075f5f646f635f5f944e8c0b5f5f636c6f737572655f5f944e8c175f636c6f75647069636b6c655f7375626d6f64756c6573945d948c0b5f5f676c6f62616c735f5f947d947****"))

$$
COMMENT '{"description": null, "pretty_name": "py_generate_series"}'
[INFO] [Aura Frame] 2026-09-30 11:28:45 User-defined function database1.py_generate_series registered successfully.
<function __main__.py_generate_series(start, endnum, step)>

步骤十：使用UDF向数据表中写入数据

  1. 在页面左上角，单击“新建单元格”。
  2. 在Notebook中的单元格，执行以下命令使用UDF向数据表中写入数据。 
    udtf_handler = con.get_function("py_generate_series", database=target_database)
ds = ds.flat_map(
    fn=udtf_handler,
    on=[ds.a, ds.b, ds.c],
    as_col='new_col'
)
ds.show()
  3. 单击左上角的，运行命令。
    命令运行后，结果示例如下则表示向数据表中写入数据成功。 
    [INFO] [Aura Frame] 2026-09-30 11:37: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 "xxx"."t1" AS "t0") AS "t1") AS "t2"
LIMIT 10
[INFO] [Aura Frame] 2026-09-30 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

步骤十一：关闭数据库连接

在使用数据库连接完成相应的数据操作后，需要关闭数据库连接。

  1. 在页面左上角，单击“新建单元格”。
  2. 在Notebook中的单元格，输入以下命令并单击左上角的，关闭数据库连接 
     con.close()

相关文档