# Exasol UDF 和脚本语言容器当用户提及 **UDF** 、 **用户定义函数** 、 **CREATE SCRIPT** 、 **ExaIterator** 、 **SCALAR** 、 **SET EMITS** 、 **BucketFS** 、 **脚本语言容器** 、 **SLC** 、 **exaslct** 、 **自定义包** 、 **GPU UDF** 、 **ctx.emit** 、 **ctx.next** 、 **variadic 时触发script** 、 **动态参数** 、 **EMITS(...)** 、 **default_output_columns** 或任何 UDF/SLC 相关主题。 ## 何时使用 UDF 使用 UDF 通过在 Exasol 集群内运行的自定义逻辑来扩展 SQL： * 每行转换（清理、解析、散列） * 跨分组行的自定义聚合 * ML模型推理（从BucketFS加载模型，评分行） * 从 SQL 内部调用外部 API * 使用 DataFrame 进行批处理 ## SCALAR 与 SET 决策指南 |标量 |设定 ---|---|--- **输入** |一次一行 |行组（通过 GROUP BY） **输出** | `RETURNS <类型>`（单值）| `EMITS (col1 TYPE, ...)`（零行或更多行） **行迭代** |不需要 |需要 `ctx.next()` 循环 **SQL 用法** | `从 t 中选择 udf(col)` | `从 t GROUP BY 键中选择 udf(col)` **用例** |每行变换 |聚合、ML 批量预测、多行发出 ## 语言选择语言 |启动 |最适合 |可通过 SLC 进行扩展吗？ ---|---|---|--- **Python 3**（3.10 或 3.12）|约 200 毫秒 |机器学习、数据科学、pandas、字符串处理 |是的 **Java**（11 或 17）| 〜1秒|企业库、类型安全、虚拟架构适配器 |是的 **Lua 5.4** | <10毫秒|低延迟转换、行级安全性 |否（本机编译到 Exasol） **R** (4.4) |约 200 毫秒 |统计建模、R模型部署|是的 ## 创建脚本语法 ### Python 标量创建或替换 PYTHON3 标量脚本 my_schema.clean_text(输入 VARCHAR(10000)) 返回 VARCHAR(10000) AS 进口再 def 运行（ctx）：如果 ctx.input 为 None：返回无 return re.sub(r'[^\w\s]', '', ctx.input).strip().lower() / 从产品中选择 clean_text(描述)； ### Python 集创建或替换 PYTHON3 设置脚本 my_schema.top_n( 项目 VARCHAR(200)，得分 DOUBLE，n INT ） EMITS（项目 VARCHAR(200)，得分 DOUBLE）AS def 运行（ctx）：行= [] 限制 = ctx.n 而真实： rows.append((ctx.item, ctx.score)) 如果不是 ctx.next(): 打破 rows.sort(key=lambda x: x[1], 反向=True) 对于项目，按行得分[:limit]： ctx.emit(项目, 分数) / SELECT top_n(产品, 收入, 5) FROM sales GROUP BY 类别; ### Java 标量创建或替换 JAVA 标量脚本 my_schema.hash_value（输入 VARCHAR(2000)）返回 VARCHAR(64) AS 导入 java.security.MessageDigest; 类 HASH_VALUE { static String run(ExaMetadata exa, ExaIterator ctx) 抛出异常 { 字符串输入 = ctx.getString("输入"); 如果（输入==空）返回空； MessageDigest md = MessageDigest.getInstance("SHA-256"); byte[] hash = md.digest(input.getBytes("UTF-8")); StringBuilder 十六进制 = new StringBuilder(); for (byte b : hash) hex.append(String.format("%02x", b)); 返回 hex.toString(); } } / ### Java 与外部 JAR 创建或替换 JAVA 标量脚本 my_schema.custom(input VARCHAR(2000)) 返回 VARCHAR(2000) AS %scriptclass com.mycompany.MyProcessor; %jar /buckets/bfsdefault/default/jars/my-lib.jar; / ### Lua 标量创建或替换 LUA 标量脚本 my_schema.my_avg(a DOUBLE, b DOUBLE) 回报双倍函数运行（ctx）如果 ctx.a == nil 或 ctx.b == nil 则返回 null end 返回 (ctx.a + ctx.b) / 2 结束 / ### R SET（机器学习预测）创建或替换 R SET SCRIPT my_schema.predict( 特征 1 双，特征 2 双） EMITS（预测双）AS 运行 <- 函数（ctx）{ 模型 <- readRDS("/buckets/bfsdefault/default/models/model.rds") 重复{ if (!ctx$next_row(1000)) 中断 df <- data.frame(f1 = ctx$feature1, f2 = ctx$feature2) ctx$emit(预测(模型, newdata = df)) } } / ## 可变参数脚本（动态参数）使用“...”接受任意数量的输入列、输出列或两者。 ### 动态输入创建或替换 PYTHON3 标量脚本 schema.to_json(...) 返回 VARCHAR(2000000) AS 导入 simplejson def 运行（ctx）：对象={} 对于范围内的 i(0, exa.meta.input_column_count, 2)： obj[ctx[i]] = ctx[i+1] # 调用者传递：名称、值、名称、值... 返回 simplejson.dumps(obj) / 从产品中选择 to_json('水果', 水果, '价格', 价格); * 按索引访问： `ctx[i]` — **Python/Java 中基于 0，Lua/R 中基于 1** * 可变参数脚本中的参数名称始终为 `0`、`1`、`2`、... — 而不是原始列名称 * `exa.meta.input_column_count` — 输入列的总数 * `exa.meta.input_columns[i].name / .sql_type` — 每列元数据 ### 动态输出 (`EMITS(...)`) 在“CREATE SCRIPT”中声明“EMITS(...)”。在调用时，必须通过以下两种方式之一提供列：方法|如有指定 |使用时 ---|---|--- **在选择中发出** |调用者的 SQL 查询 |输出结构取决于数据值 **`default_output_columns()`** |脚本正文 |输出结构可仅从输入列计数/类型导出 -- SELECT 中的 EMITS（当输出取决于数据内容时需要） SELECT split_csv(line) EMITS (a VARCHAR(100), b VARCHAR(100), c VARCHAR(100)) FROM t; # default_output_columns() — 在 run() 之前调用，没有可用的 ctx/数据访问 def default_output_columns(): 零件=[] 对于范围内的 i(exa.meta.input_column_count)： parts.append("c" + exa.meta.input_columns[i].name + " " + exa.meta.input_columns[i].sql_type) 返回“,”.join(parts) 如果两者均未提供，则查询将失败并显示： > _脚本具有动态返回参数。通过 EMITS 指定查询中的返回参数或在 UDF 中实现 default_output_columns 方法。_ ## ExaIterator API 快速参考 ###Python 方法/属性 |标量 |设置|描述 ---|---|---|--- `ctx.<列>` |是的 |是的 |访问输入列值 `返回值` |是的 |没有|返回单个值（RETURNS） `ctx.emit(v1, v2, ...)` |没有|是的 |发射输出行 (EMITS) `ctx.emit(数据帧)` |没有|是的 |将 DataFrame 作为行发出 `ctx.next()` |没有|是的 |前进到下一行；最后返回“False” `ctx.size()` |没有|是的 |当前组中的行数 `ctx.reset()` |没有|是的 |将迭代器重置为第一行 `ctx.get_dataframe(num_rows, start_col)` |没有|是的 |获取行作为 pandas DataFrame **重要：** 没有 `emit_dataframe()` 方法 — 使用 `ctx.emit(dataframe)` 来发出 DataFrame。 ### Java 方法|描述 ---|--- `ctx.getString("col")` |获取字符串值 `ctx.getInteger("col")` |获取整数值 `ctx.getDouble("col")` |获得双倍值 `ctx.getBigDecimal("col")` |获取十进制值 `ctx.getDate("col")` |获取日期值 `ctx.getTimestamp("col")` |获取时间戳值 `ctx.next()` |前进到下一行（仅限 SET） `ctx.emit(v1, v2, ...)` |发出输出行（仅限 SET） `ctx.size()` |组中的行数（仅限 SET） `ctx.reset()` |重置到第一行（仅限 SET） ## BucketFS 文件访问所有语言都可以从 BucketFS 读取文件，路径为 `/buckets///`： # Python — 加载 pickled ML 模型进口泡菜将 open('/buckets/bfsdefault/default/models/model.pkl', 'rb') 作为 f：模型 = pickle.load(f) // Java — 通过 %jar 指令引用 JAR %jar /buckets/bfsdefault/default/jars/my-library.jar; **性能提示：** 加载模型/资源一次（在行循环外部或模块级变量中），而不是每行。 ## GPU 加速（Exasol 2025.2+） Exasol 通过支持 CUDA 的脚本语言容器支持 GPU 加速的 UDF： * 使用 `template-Exasol-8-python-3.{10,12}-cuda-conda` 风格 * Exasol 主机上需要 NVIDIA 驱动程序 * 通过 SLC 中的 conda 安装 GPU 库（PyTorch、TensorFlow、RAPIDS） * 标准 UDF API — 除了导入 GPU 库之外无需更改任何代码 ## 脚本语言容器 (SLC) 概述 UDF 在脚本语言容器（基于 Docker 的运行时环境）内运行。默认的 SLC 包含标准库。当您需要其他包（例如 scikit-learn、PyTorch、自定义 JAR）时，构建自定义 SLC。 ### 当您需要定制 SLC 时 * 安装不在默认容器中的 pip/conda 包 * 添加系统库（apt包） * 使用不同的 Python 版本（3.10 与 3.12） * 启用 GPU/CUDA 支持 * 从 CRAN 添加 R 包 ### 快速激活 -- 激活当前会话更改会话集 SCRIPT_LANGUAGES='PYTHON3=localzmq+protobuf://////?lang=python#buckets/////exaudf/exaudfclient_py3'; -- 激活系统范围（需要管理员）更改系统设置 SCRIPT_LANGUAGES='...'; ### 安装构建工具 pip 安装 exasol-脚本语言-容器工具 ## 性能提示 * **加载一次，使用多次** ：在行循环之外加载模型/资源 * **使用SET进行批处理** ：将行收集到列表/DataFrame中，批量处理 * **Lua 实现低延迟**：避免 JVM/Python 启动开销 * **并行性是自动的**：UDF 同时在所有集群节点上运行 ## 详细参考资料 * **Python 模式** — 上下文 API、DataFrame 模式、类型映射、测试：[references/udf-python.md]() * **Java 和 Lua 模式** — ExaMetadata API、JAR、适配器、Lua 库：[references/udf-java-lua.md]() * **构建自定义 SLC** — exaslct CLI、风格、自定义、部署、故障排除：[references/slc-reference.md]()

Exasol Udfs

包含技能