Aura DataFrame SDK新手入门

流程概览

步骤1：获取并安装SDK安装包

1. 请联系技术支持获取SDK安装包。
2. 安装SDK。
   1. 下载并解压SDK安装包。
   2. 打开命令行窗口，进入解压目录。
   3. 执行如下命令安装SDK，安装过程中会自动下载第三方依赖库。

      pip install huawei_aura_frame-x.x.x-py3-none-any.whl

      x.x.x为具体版本号，以实际为准。

步骤2：建立连接

通过ai_lake.connect() 创建与 Aura 端点的会话连接；使用UDF前还需配置部署工作区。

详细操作请参考建立与AuraJobV2端点的连接。

创建会话，配置连接信息（Endpoint、工作空间ID、Catalog、AK/SK、数据库等）：

import os
import logging
from aura_frame.multimodal import ai_lake

con = ai_lake.connect(
    aura_endpoint=os.getenv("aura_endpoint"),           # AuraJobV2端点所在的终端节点（Endpoint）
    aura_endpoint_id=os.getenv("aura_endpoint_id"),      # AuraJobV2端点ID
    aura_endpoint_name=os.getenv("aura_endpoint_name"),  # AuraJobV2端点名称
    aura_workspace_id=os.getenv("aura_workspace_id"),    # 工作空间ID
    lf_catalog_name=os.getenv("lf_catalog_name"),        # LakeFormation Catalog名称
    lf_instance_id=os.getenv("lf_instance_id"),          # LakeFormation实例ID
    project_id=os.getenv("project_id"),                  # 项目ID
    access_key=os.getenv("access_key"),                  # 访问密钥ID（AK）
    secret_key=os.getenv("secret_key"),                  # 访问密钥（SK）
    default_database="postgres",                         # 默认数据库
    use_single_cn_mode=True,                             # 是否开启单CN模式
    logging_level=logging.INFO,                          # 设置日志级别
)

步骤3：准备数据

创建数据表、写入数据、加载已有数据集。

1. 新创建表和数据。
   1. 在LakeFormation中创建数据库，例如：database1。具体操作，请参见创建LakeFormation实例并规划元数据。
   2. 创建数据表（库中无目标表时）。
      create_table支持通过store指定表格式（ORC/PARQUET/ICEBERG等）、通过partition_by指定分区列，详见create_table。

      import ibis
      table_name = "table_name"
      table_schema = ibis.schema({"col_a": "int32", "col_b": "string"})
      con.sql(f'DROP TABLE IF EXISTS database1."{table_name}";')
      con.create_table(
          table_name,
          schema=table_schema,
          database="database1",
          table_format="orc",
          if_not_exists=True,
      )

   3. 通过SQL，写入数据。

      con.sql("INSERT INTO database1.table_name VALUES(1, 's1'), (2, 's2')")

      详细操作请参考创建表并写入数据。

2. 加载已有数据集。

   ds = con.load_dataset("table_name", database="database1")  # 从指定数据库加载表，建立表对象，table_name和database1需要替换为实际的表名和数据库名称

步骤4：数据处理

使用DataFrame API对数据进行筛选、聚合、排序、连接等加工。

DataFrame 操作采用懒加载机制，以下操作只是“记录”计算意图，调用execute()才真正在Aura引擎执行并返回Pandas DataFrame。

详细操作请参考数据加工与分析。

# 选择列
ds=ds.select_columns([ds.col_a, ds.col_b])
# 筛选行
ds=ds.filter(ds.col_a > 0)
# 分组聚合（对数值列col_a做聚合）
ds=ds.groupby("col_a").aggregate(sum_b=ds.col_a.sum(), avg_b=ds.col_a.mean())
# 排序
ds=ds.order_by(ds.col_a.desc())
# 触发执行，返回Pandas DataFrame（关键一步）
df=ds.execute()
print(df)

步骤5：更多数据开发操作

根据业务需要，进一步使用UDF与多模态处理能力。

• 注册和使用Scalar UDF/Class UDF/UDAF ，请参考使用Scalar UDF处理数据、使用Class UDF处理数据、使用聚合统计的UDAF
• UDF类型、注册方式、开发规范，请参考用户自定义函数UDF。

步骤6：查看作业运行结果

作业运行成功后，运行日志会回显执行过程与结果（例如建表、执行SQL、返回查询结果等），可据此验证作业是否按预期运行。

更多示例参考使用Aura DataFrame SDK提交作业。

数据操作完成后关闭会话释放资源：

con.close()  # 关闭会话，释放资源