使用AuraJobV2运行示例

前提条件

• 已创建OBS并行文件系统，用于存放SDK上传的UDF包。具体操作，请参见创建并行文件系统。
• 已创建LakeFormation实例，并确认实例下存在数据目录（Catalog）与数据库。具体操作，请参见创建LakeFormation实例并规划元数据。
• 已在LakeFormation中注册数据表（本示例至少包含id、value两列）。
• 已创建AuraJobV2端点，并获取端点名称、工作空间ID等信息。具体操作，请参见创建Aura引擎端点。
• 已获取AK和SK信息，具备OBS并行文件系统和LakeFormation访问权限。
• 已获取并安装Aura DataFrame SDK。具体操作，请参见Aura DataFrame SDK获取与安装。

示例

# 1. 导入依赖
import os
from aidatalake_catalog import Parameter, Signature
from aidatalake_catalog.client import AIDatalake_Catalog_Client
from aura_frame.multimodal import ai_lake

# 2. 定义带类型注解的UDF
class Multiplier:
    """Class 定义 UDF：乘以可配置 factor（默认 2）。"""
    def __init__(self, factor: int = 2):
        self._factor = factor
    def __call__(self, x: int) -> int:
        return x * self._factor

# 3. 定义函数签名（可选，未传入时由SDK根据类型注解自动推断）
sig = Signature(
    parameters=[Parameter(name="x", annotation=int)],
    return_annotation=int,
)
# 4. 初始化Catalog客户端（AK/SK建议从环境变量读取，避免硬编码）
client = AIDatalake_Catalog_Client(
    ak=os.getenv("XXX_SDK_AK"),
    sk=os.getenv("XXX_SDK_SK"),
    security_token=None,                       # 临时凭证时使用
    obs_endpoint="obs.cn-north-4.myhuaweicloud.com",
    lakeformation_endpoint="https://lakeformation.cn-north-4.myhuaweicloud.com",
    lakeformation_instanceid="your-instance-id",
    catalog_name="your-catalog-name",
    default_database="your_default_db",
    obs_path="your-bucket/udf/",              # bucket_name/dir/subdir
    project_id="your-project-id",
    ignore_ssl_verification=False,
)

# 5. 注册自定义函数
func = client.create_udf(
    Multiplier,
    name="class_multiplier_def",
    signature=sig,
    description="class 定义 UDF: 乘以 factor (默认 2)",
)

# 6.使用Aura引擎执行函数
"""
|  id  | value|
|------|------|
|  1   |  10  |
|  2   |  20  |
|  3   |  30  |
"""
con = ai_lake.connect(
        aura_endpoint=os.getenv("aura_endpoint"),
        aura_endpoint_name=os.getenv("aura_endpoint_name"),
        aura_workspace_id=os.getenv("aura_workspace_id"),
        lf_catalog_name=os.getenv("lf_catalog_name"),
        access_key=os.getenv("access_key"),
        secret_key=os.getenv("secret_key"),
        default_database=target_database,
        use_single_cn_mode=True
)

ds = con.load_dataset("table_1")  # table_1需要提前在LakeFormation注册
ds = ds.map(fn=f'{database}.class_multiplier_def', on=['value'], as_col='mul_result')
df = ds.execute().sort_values('id').reset_index(drop=True)

print(f"  {'ID':<5} {'原始值':<10} {'mul_result':<15}")
print("-" * 35)
for _, r in df.iterrows():
    print(f"{int(r['id']):<5} {int(r['value']):<10} {int(r['mul_result']):<15}")