# 使用Class UDF处理数据
Class UDF基于面向对象编程范式实现，适合需要初始化参数、加载资源或有状态的处理逻辑。通过__init__方法初始化资源（如加载模型），__call__方法对每行数据执行计算，可按需通过__del__方法清理资源。
#### 场景描述
在文本处理场景中，需要使用SentencePiece分词模型对text_samples表中的文本列content进行子词切分（subword tokenization）。模型文件需要在UDF初始化时一次性加载，之后对每行文本重复使用。通过Class UDF，可以在__init__中加载spiece.model模型文件并配置EOS编码选项，在__call__中对每行文本执行分词并返回JSON格式的分词结果，避免逐行重复加载模型的开销。
#### 约束与限制
- 类必须定义__init__方法，且其参数名称需与调用时的参数名称保持一致。
- 类必须定义 __call__ 成员方法，该方法为UDF的主函数入口。
- 类可按需定义 __del__ 成员方法，且该方法不支持传入参数。
- 调用Class UDF时，__init__ 方法的参数仅支持传入常量，不支持传入表的数据列或其他表达式。
 
#### 前提条件
- 已创建表并写入数据。具体操作，请参见[创建表并写入数据](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_09_0030.html)。
- 运行时环境要求Python版本为3.11。
 
#### 准备工作
本示例使用SentencePiece分词模型，需要准备模型文件spiece.model。推荐从Hugging Face获取T5模型自带的SentencePiece分词器，获取方式如下：
1. 安装SentencePiece依赖：
   ```
   pip install sentencepiece==0.2.0
   ```
   
2. 安装huggingface命令行工具：
   ```
   pip install huggingface-hub
   ```
   
3. 下载模型文件到示例脚本同级目录：
   ```
   hf download google-t5/t5-small spiece.model --local-dir .
   ```
   
 
如果hf download命令无法使用，可通过浏览器手动下载：
- 在Hugging Face网站搜索google-t5/t5-small，进入模型主页。
- 单击"Files and versions"标签页。
- 找到spiece.model文件，单击文件名进入详情页。
- 单击文件内容右上方的下载按钮，将文件保存到本地。
- 将下载的spiece.model文件放置在与示例脚本同级目录下。
#### 示例
本示例定义一个基于SentencePiece分词模型的Class UDF，在__init__中加载模型文件，在__call__中对每行文本执行子词切分，将其注册到数据库后通过map方法处理text_samples表的文本列。
1. 定义Class UDF类SPManager，在__init__中加载SentencePiece模型文件，在__call__中对文本执行分词并返回JSON格式结果。
2. 连接AuraJobV2端点并设置UDF工作空间。
3. 删除已存在的同名UDF（支持重复执行）。
4. 注册Class UDF到目标数据库，指定依赖包和模型文件。
5. 获取已注册的UDF算子并加载text_samples数据集。
6. 通过map方法调用UDF对content列执行分词，生成pieces_column列。
7. 选择content和pieces_column列并执行查询打印结果。
8. 关闭AuraJobV2端点连接。
```
import os
from aura_frame.multimodal import ai_lake
from sentencepiece import SentencePieceProcessor
target_database = "test"  #在LakeFormation中创建数据库，例如：test。具体操作，请参见创建LakeFormation实例并规划元数据
#连接AuraJobV2端点，具体参数说明，请参见建立与AuraJobV2端点的连接。
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名称
    access_key=os.getenv("access_key"),                # 访问密钥ID（AK）
    secret_key=os.getenv("secret_key"),                # 访问密钥（SK）
    default_database=target_database,                       # LakeFormation数据库名称
    use_single_cn_mode=True,   # 是否开启单CN模式
    logging_level=logging.INFO # 设置日志级别
)
#配置UDF部署工作区，具体参数说明，请参见配置UDF部署工作区。
con.set_function_staging_workspace(
    obs_directory_base=os.getenv("obs_directory_base"),  # 并行文件系统中UDF的存储路径
    obs_bucket_name=os.getenv("obs_bucket_name"),       # 并行文件系统名称
    obs_server=os.getenv("obs_server"),                  # OBS终端节点（Endpoint）
    access_key=os.getenv("access_key"),                  # 访问密钥ID（AK）
    secret_key=os.getenv("secret_key"),                  # 访问密钥（SK）
)
class SPManager:
    def __init__(self, model_file: str, *, bos: bool = False, eos: bool = False, reverse: bool = False) -> None:
        sp = SentencePieceProcessor(model_file=model_file)
        opts = ":".join(opt for opt, flag in (("bos", bos), ("eos", eos), ("reverse", reverse)) if flag)
        sp.set_encode_extra_options(opts)
        self.spm = sp
    def __call__(self, input_text: str) -> str:
        import json
        return json.dumps(self.spm.encode(input_text, out_type=str))
try:
    # 删除已存在的同名UDF（支持重复执行）
    con.delete_function("SPManager", database=target_database, if_it_exists=True)
    # 注册Class UDF
    con.create_scalar_function(
        SPManager,
        database=target_database,
        packages=["sentencepiece==0.2.0"],
        imports=["spiece.model"],
    )
    # 获取UDF算子并在DataFrame中使用
    udf = con.get_function("SPManager", database=target_database)
    ds = con.load_dataset("text_samples", database=target_database)
    ds = ds.map(
        fn=udf, on=[ds.content], as_col="pieces_column",
        model_file="spiece.model", eos=True,
    )
    ds = ds.select_columns(ds.content, ds.pieces_column)
    df = ds.execute()
    print(df)
finally:
    con.close()
```
#### 相关文档
- [Session APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0135.html)
- [Dataset APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0132.html)
- [Table APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0133.html)
- [User-Defined Function APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0134.html)