byzer 笔记总结

1.总览（简单了解）

1.1 数据挖掘的定义

基于大数据技术，针对有价值是业务场景，对数据中台沉淀的大量数据进行探索，分析。寻找数据与数据之间潜藏的关系，转化为自动化的算法模型，从而获取有价值的规律和标签，提升业务能力。

数据挖掘是一个多学科交叉的产物，涉及统计学、数据库、机器学习、人工智能及模式识别等多种学科，如图所示。

数据挖掘方法分类介绍

数据挖掘方法按照来源进行分类显得过于庞杂，而且不便于理解和记忆。按照其目的，将数据挖掘方法分为预测性和描述性两大类，如下所示。

目的：预测性
- 定义：有监督学习，分类模型，用一个或多个自变量预测因变量的值 举例：客户是否会违约是一个因变量，可以根据客户的性别、年龄、收入、职位、经济状况、历史信用状况等因素进行预测
- 主要算法：决策树、线性回归、逻辑回归、支持向量机、神经网络、判别分析等
目的：描述性
- 定义：无监督学习，分析具有多个属性的数据集，找出潜在的模式，没有因变量
- 举例：观察个体之间的相似程度，如根据年龄、性别、收入等因素进行客户细分。根据客户对多个产品的购买情况发现产品之间的相关性
- 主要算法：聚类、关联分析、因子分析、主成分分析、社交网络分析等

1.2.数据挖掘方法论

数据挖掘为什么需要方法论：

数据挖掘方法论为开展数据挖掘项目提供了一套完整的、高效的、质量可控的项目管理过程；

随着数据挖掘市场的发展和成熟，由不同的组织机构提出过很多的方法论，如CRISP-DM、SEMMA、5A等，其中CRISP-DM、SEMMA是应用最为广泛。

SEMMA：包括抽样（Sample）、探索（Explore）、修订（Modify）、建模（Model）和评估（Assess）。

CRISP-DM：包括业务理解(business understanding），数据理解 (data understanding），数据准备(data preparation)，建立模型(modeling），评估模型(evaluation)和结果部署(deployment)。

CRISP-DM 更偏向于业务场景交付的理论，SEMMA更偏向算法科学开发的理论；

长城更适用CRISP-DM方法论，与业务联合共创的同时，减少业务与IT之间的gap，降低数据挖掘项目风险，将研发投入与业务需求放在一起形成闭环管理。并沉淀方法论、工具、算法、流程、人员培养。最大化的复用数据模型、算法模型、数据分析SOP流程等等，逐渐提升交付效率，降低实施成本。

CRISP-DM (cross-industry standard process for data mining)

生命周期包括六个阶段：

业务理解（business understanding）
数据理解（data understanding）
数据准备（data preparation）
建模（modeling）
评估（evaluation）
部署（deployment）

1.2.1.理解业务需求

业务需求决定了分析的方向，明确业务需求和分析目标

明确what are you doing,why you are doing

①确定业务目标
②评估情况【清理数据源来历】【需求、约束及风险】

1.2.2.数据理解【理解业务，探索业务需求中的指标概念和影响因素】

明确需要哪些数据，并明确地定义数据【格式和场景】

这一步的作用：帮助自己了解哪些数据可用，使业务需求和实际的数据保持一致性，验证业务需求是否可行，并细化实现指标和内容

①收集原始数据通过【数据在项目资源中的列表】，明确【数据的位置和获取数据的方法】
②描述数据明确【数据表面特性、数据格式和质量】充分利用数据字典
③探索数据重点把握【数据整体趋势及特殊子集】
④验证数据质量验证内容【数据全，正确与否、是否有噪声数据或异常值】等

方法：

通过数据字典、业务需求的数据定位、额外数据集、构建数据（原因、方法和规则）、用可视化方式对数据进行探索、评估数据质量

基于选择的数据和业务需求选择分析方法、格式化数据、进行分析

1.2.3.数据准备【业务数据与分析数据格式不同，需要做转换】

①选择数据【考虑与业务需求的相关性、数据质量和技术约束等因素】
②清洗数据【通过选择、替换等方法提高数据质量】
③构造数据【构造衍生属性】
④集成数据【同源数据合并和不同源数据合并】
⑤格式化数据【根据业务需求对数据进行格式化】

1.2.4.建模【选择分析技术对数据分析计划进行模块化】

①选择建模技术【业务理解阶段，用到一些算法模型】
②生成测试设计【分离测试数据和训练数据，定义模型结果验证参数】
③建立模型【列出参数和选择值，评估模型】

1.2.5. 评估【从业务角度评估结果】

①从业务角度评估结果
②审核过程【是否有重要的因素被忽略】
③确定下一步【验证可行然后部署实施】

1.2.6. 部署【实现数据分析应用到业务中】

①计划实施【确定如何使用分析及挖掘结果来达到业务需求的目标】
②计划的监控和维护【数据分析实施的计划应用到业务系统中，数据和结果反馈】
③最终的报告
④项目回顾总结

过程中出具的工作结果：

【出具的内容】

【分析问题的思维导图、明确项目依据和目标（业务理解和指标细化）、根据数据字典进行数据分析、总结性的数据探索报告、分析计划（维度和度量等指标确定和实现）、最终验证报告、业务实现】

1.3. MLSQL数据挖掘新一代的编程语言

MLSQL（Machine Leaning SQL）是一门全新的编程语言，完全开源，低代码，使用 MLSQL可以实现数据处理、数据分析和 AI。

我们使用一种语言 MLSQL，就可以在单一平台上实现过去要使用多语言、多平台、多组件才能实现的事情。

MLSQL可以实现使用统一的交互语言，在一个统一的平台内即可完成数据处理 +数据分析+数据科学的任务：

MLSQL主要适用以下两种人群：

因为数据科学家人员稀少，储备成本高昂，MLSQL最大的优势就是降低对数据科学家的依赖，让数据工程师和数据分析师也可以做基础的数据挖掘工作。

MLSQL语言架构

作为一个解释型语言，拥有解释器（Interpreter）以及运行时环境（Runtime），Byzer 的 Interpreter 会对 Byzer 的脚本做词法分析，预处理，解析等，然后生成 Java/Scala/Python/SQL 等代码，最后提交到 Runtime 上进行执行

MLSQL使用 Github 作为包管理器（Package Manager），有内置 lib 库和第三方 lib 库（lib 库是使用 Byzer 语言编写出的功能模块）

MLSQL 既保留了 SQL 的优势，简洁易懂，还允许用户通过扩展点来进行更多的进阶操作，提供更多可编程能力，提供了变量成员（Variable），函数成员（Macro Function），以及提供了模块（Lib）以及包（Package）的设计。开发者可以通过 Macro Function 来封装代码功能，或实现 Byzer 的语法糖来增强语法特性；也可以通过 Package 进行代码组织，开发可供第三方使用的功能库。

1.4. MLSQL语言的扩展架构

包括：应用生态扩展点、语言层扩展点、内核层扩展点。

1.4.1 MLSQL 应用生态扩展

这个层面主要是面向使用 MLSQL ，具备 Scala/Java 开发能力的工程师。通过他可以增加 MLSQL 处理业务需求的能力。大致分为下面几类：

数据源扩展（数据源扩展从接入数据和消费数据两个角度来进行区分）
- 接入数据，将数据源抽象成 MLSQL 内的表
- 消费数据，将 MLSQL 处理过的数据通过 API 进行暴露，给下游进行消费
ET 扩展（Estimator-Transformer）：ET 定义了对表操作的扩展接口，常见的通过 ET 接口实现的一些扩展如
- UDF 函数
- Byzer-Python，拥抱 Python 生态来扩展 Byzer 处理表的能力
- !plugin 插件及插件管理
- Macro Function 实际上也是 ET 的一个实现
App 扩展：通过 Byzer 的接口来实现应用功能的扩展
- 比如通过实现 LSP 协议，暴露 auto-suggest 接口来完成 Byzer 的代码提示

1.4.2 MLSQL 语言层扩展

该层扩展主要供 Byzer 编码者使用来扩展语言能力，比如编写 Byzer 的第三方功能库，常见的扩展点如下：

Macro Function （宏函数）在 Byzer-Lang 中就是函数成员，类似 Method in Java。
UDF 函数 (允许用户编写 Scala/Java UDF 函数，作用于 select / 分支条件表达式中)
Byzer-python ( 作用于 run 语句中。支持 Python 脚本 )

这里着重提一下 Byzer-python，通过 Byzer-python 的扩展，用户就可以在 Byzer 中拥抱 Python 的生态了。

对于一些资深的机器学习用户，他们熟悉例如在 Jupyter Notebook 中用 Python 的 scikit-learn 或者 tensorflow 进行了机器学习的模型开发，在 Byzer 中我们通过 Byzer-python 来实现对 Python 代码的引用和适配，资深 Python 用户可以继续使用您习惯的机器学习的包进行模型开发。这里引用了Byzer 的例子，利用 Byzer-python 能力，实现分布式图片处理。

#%python#%input=raw_cifar10_table#%output=cifar10_resize#%cache=true#%schema=st(field(content,binary),field(path,string))#%dataMode=data#%env=source /opt/miniconda3/bin/activate ray1.8.0import io,cv2,numpy as np from pyjava.api.mlsql import RayContext ray_context = RayContext.connect(globals(),"127.0.0.1:10001")def resize_image(row): new_row = {} image_bin = row["content"] oriimg = cv2.imdecode(np.frombuffer(io.BytesIO(image_bin).getbuffer(),np.uint8),1) newimage = cv2.resize(oriimg,(28,28)) is_success, buffer = cv2.imencode(".png", newimage) io_buf = io.BytesIO(buffer) new_row["content"]=io_buf.getvalue() new_row["path"]= row["path"] return new_row ray_context.foreach(resize_image)

1.4.3 MLSQL 内核层扩展：

这个层面的扩展点主要面向 Byzer-lang 的核心开发者的。通过这些扩展点，可以增强引擎的能力。本质上其实是增强 Byzer-Lang 的解释器

内核生命周期扩展点来实现插件扩展，这里的生命周期则是指在 Runtime 初始化之前还是之后，比如 Byzer CLI，就是在 Runtime 初始化之前完成插件加载的
新语法扩展点，比如新增一个语法关键字
各种自定义功能扩展点，比如权限，接口访问控制，异常处理，请求清理等。

1.5 分布式技术架构

MLSQL 执行引擎是分布式架构，基于云原生架构设计，用户可以使用桌面版软件和webUI连接到云端引擎，轻松解锁算力和存储空间限制。

1.6 机器学习数据库

1.6.1.OpenMLDB

OpenMLDB 提供生产级 FeatureOps 全栈解决方案

OpenMLDB 是一个提供线上线下一致性的生产级特征平台，我们对外提供的是一整套的 SQL 语言。用户可以通过 SQL 语言写成脚本，再用 OpenMLDB 离线引擎做批量计算，进行模型探索。探索完成后，SQL 脚本能直接上线通过 OpenMLDB 的在线实时引擎完成实时特征计算

OpenMLDB 在线架构的主要模块有 Apache ZooKeeper, nameserver 以及 tablet集群（一个tablet进一步包含了SQL engine 和 storage engine）。下图显示了这些模块之间的关系。其中 tablets 是整个 OpenMLDB 存储和计算的核心模块，也是消耗资源最多的模块。ZooKeeper 和 nameserver 主要用于辅助功能，如元数据的管理和高可用等。

1.6.2.Tablets

OpenMLDB 使用Tablet模块执行SQL、存储数据。该模块是整个 OpenMLDB 功能实现的核心以及资源占用的瓶颈。从功能上看，Tablet进一步包含了 SQL engine 和 storage engine 两个模块。Tablet 也是 OpenMLDB 部署资源时可调配的最小单元。一个 tablet 不能被拆分到多个物理节点；但是一个物理节点上可以有多个 tablets。

1.6.3.SQL Engine

SQL engine 负责执行 SQL 查询计算。SQL engine 收到 SQL 查询的请求后的执行过程如下图所示：

SQL 引擎通过 ZetaSQL 把 SQL 解析成AST语法树。因为我们加入了 LAST JOIN，WINDOW UNION 等专门针对特征工程设计的特殊 SQL 语法，所以对开源的 ZetaSQL 做了优化。经过如上图一系列的编译转化、优化，以及基于 LLVM 的 codegen 之后，最终生成执行计划。SQL 引擎基于执行计划，通过 catalog 获取存储层数据做最终的 SQL 执行运算。在分布式版本中，会生成分布式的执行计划，会把执行任务发到其他 tablet 节点上执行。目前 SQL 引擎采用 push 的模式，将任务分发到数据所在的节点执行，而不是将数据拉回来。这样做的好处可以减少数据传输。

1.6.4.Storage Engine

Storage engine 负责 OpenMLDB 数据的存储，以及支持相应的高可用相关的功能。

Note

OpenMLDB 支持基于内存或者磁盘的两种存储引擎，本文针对自研的内存存储引擎进行介绍。磁盘存储引擎基于 RocksDB，原理上也保持一致，可以参考 RocksDB。

1.6.5.数据分布

OpenMLDB 集群版是一个分布式的数据库，一张表的数据会进行分片，并且建立多个副本，最终分布在不同的节点中。这里展开说明两个重要的概念：副本和分片。

副本（replication）：为了保证高可用以及提升分布式查询的效率，数据表将会被存放多个拷贝，这些拷贝就叫做副本。
分片（partition）：一张表（或者具体为一个副本）在具体存储时，会进一步被切割为多个分片用于分布式计算。分片数量可以在创建表时指定，但是一旦创建好，分片数就不能动态修改了。分片是存储引擎主从同步以及扩缩容的最小单位。一个分片可以灵活的在不同的 tablet 之间实现迁移。同时一个表的不同分片可以并行计算，提升分布式计算的性能。OpenMLDB 会自动尽量使得每一个 tablet 上的分片数目尽量平衡，以提升系统的整体性能。一张表的多个分片可能会分布在不同 tablet 上，分片的角色分为主分片（leader）和从分片（follower）。当获得计算请求时，请求将会被发送到数据对应的主分片上进行计算；从分片用于保证高可用性。

如下图显示了一个数据表，在两个副本的情况下，基于四个分片，在三个 tablets 上的存储布局。实际使用中，如果某一个或者几个 tablet 的负载过高，可以基于分片，进行数据迁移，来改善系统的负载平衡和整体的吞吐。

1.6.6.数据持久化及主从同步

目前版本的 OpenMLDB 的在线数据全部保存在内存中，为了实现高可用会把数据通过 binlog 以及 snapshot 的形式持久化到硬盘中。

如上图所示，服务端收到 SDK 的写请求后会同时写内存和 binlog。binlog 是用来做主从同步的，数据写到 binlog 后会有一个后台线程异步的把数据从 binlog 中读出来然后同步到从节点中。从节点收到同步请求后同样进行写内存和 binlog操作。Snapshot 可以看作是内存数据的一个镜像，不过出于性能考虑，snapshot 并不是从内存 dump 出来，而是由 binlog 和上一个 snapshot 合并生成。在合并的过程中会删除掉过期的数据。OpenMLDB会记录主从同步和合并到 snapshot 中的 offset, 如果一个 binlog 文件中的数据全部被同步到从节点并且也合并到了 snapshot 中，这个 binlog 文件就会被后台线程删除。

OpenMLDB 完成从特征的离线开发到上线部署，只需要以下三个步骤：

1.使用 SQL 进行离线特征脚本开发，用于模型训练。
2.SQL 特征脚本一键部署上线，由线下模式切换为线上模式。
3.接入实时数据，进行线上实时特征计算，用于模型推理。

在离线部分，OpenMLDB 的离线特征计算引擎是基于 Spark 做了一个改造。Spark 会用 JNI 的方式来调用我们生成的SQL解析执行库。在线部分，我们用自研实时计算引擎来做实时计算。OpenMLDB离线和在线引擎使用同一套一致性执行计划生成器，运行同一套代码，天然保证了线上线下的一致性。

功能

以 SQL 为核心的开发和管理体验低门槛且功能强大的数据库开发体验，全流程基于 SQL 进行特征计算脚本开发以及部署上线
线上线下一致性执行引擎离线和实时特征计算使用统一的计算执行引擎，线上线下一致性得到了天然保证
面向特征计算的定制化优化离线特征计算提供基于源代码优化的Spark高性能版本，线上实时特征计算在高吞吐压力下的复杂查询提供几十毫秒量级的延迟，充分满足高并发、低延迟的性能需求
生产级特性为大规模企业应用而设计，整合诸多生产级特性，包括灾备恢复、高可用、可无缝扩缩容、可平滑升级、可监控、异构内存架构支持等

特点

– 线上线下一致性：离线和实时特征计算引擎使用统一的执行计划生成器，线上线下计算一致性得到了天然的保证。

– 毫秒级超低延迟的实时 SQL 引擎：线上实时 SQL 引擎基于完全自研的高性能时序数据库，对于实时特征计算可以达到毫秒级别的延迟，性能远超流行商业内存数据库（可参考 VLDB 2021 上的论文），充分满足高并发、低延迟的实时计算性能需求。

– 基于 SQL 定义特征：基于 SQL 进行特征定义和管理，并且针对特征计算，对标准 SQL 进行了增强，引入了诸如 LAST JOIN 和 WINDOW UNION 等定制化语法和功能扩充。

– 生产级特性：为大规模企业应用而设计，整合诸多生产级特性，包括分布式存储和计算、灾备恢复、高可用、可无缝扩缩容、可平滑升级、可监控、异构内存架构支持等。

2. 快速开始（需要简单实操）

2.1. Notebook （MSSQL IDE 开发工具）

Notebook作为MLSQL 的Web前端开发IDE工具，方便用户开发和使用，完成诸如高亮，代码提示，运行等能力。同时也是 MLSQL 的 debugger(调试器)，通过 Notebook 简单易用的调试功能，省去了初学者安装配置开发环境的门槛。

目前不仅支持官方 Web 版本 IDE：Byzer Notebook ，也支持用户在 VSCode 中下载插件来启用代码编辑以及 Notebook 的能力。

2.2.注册登录

访问地址：http://10.255.132.9:9002

使用前先进行用户注册：

2.3.界面功能介绍

2.3.1.工作区介绍

本章将详细介绍 Byzer Notebook 工作区相关操作。您可以在页面顶栏点击工作区进入工作区页面，创建或编辑笔记本。

2.3.2.工作区组织形式

在工作区左侧导航栏中，我们以文件的形式组织笔记本和工作流，您可以创建不同的主题或用户的文件夹及子文件夹，然后将笔记本或工作流放置在文件夹下。

工作区右侧是笔记本或工作流的编辑页面，您可以在左侧导航栏中打开笔记本或工作流进行编辑。

2.3.3.创建/导入文件夹、笔记本、工作流

您可以点击左上角工作区右侧的 + 按钮创建文件夹、笔记本、工作流，或者导入笔记本或工作流文件。您也可以点击工作区中已有的文件夹右侧浮标 ···，直接在文件夹下面创建子文件夹、笔记本、工作流。

2.3.4.移动文件夹、笔记本、工作流

创建和导入的文件夹、笔记本或工作流都默认在根目录下，您可以在左侧导航栏的工作区模块中找到对应的项目，将鼠标悬浮在对应项目名称上，并点击 ··· -> 移动将他们移动到相应的文件夹下。

重命名文件夹、笔记本、工作流

您可以点击文件夹、笔记本、工作流右侧的 ··· -> 重命名按钮给他们重新命名。

克隆文件夹、笔记本、工作流

您可以点击文件夹、笔记本、工作流右侧的 ··· -> 克隆按钮克隆一个完全一样的文件夹、笔记本或工作流。克隆文件夹时将在同级目录克隆一个包含所有子文件的文件夹。

导出笔记本、工作流

您可以点击笔记本或工作流右侧的 ··· -> 导出按钮导出笔记本或工作流。

删除文件夹、笔记本、工作流

您可以点击文件夹、笔记本、工作流右侧的 ··· -> 删除 按钮删除文件夹、笔记本或工作流，注意删除之后无法恢复，请谨慎操作。

注意：文件夹下没有内容时才可以被删除。

2.3.5 notebook 使用和编辑

点击顶栏的工作区进入工作区编辑页面，在左侧导航栏中的工作区模块中选择您想要操作的笔记本。在笔记本的上方，有一排操作按钮，如下图所示：

他们的操作含义分别是：

新建 / 克隆 / 删除 / 重命名笔记本：可以在当前笔记本主界面下创建新的笔记本，也可以克隆、重命名和删除当前笔记本；

保存：保存当前编辑；

新增单元格：在笔记本内新增一个单元格；

选择语言：选择单元格内的语言，支持 Byzer、Python、Markdown 语言；

运行：运行光标所在的单元格；

运行下一个：运行光标所在的单元格的下一个单元格；

运行所有：运行笔记本内的所有单元格,运行时会展示运行的进度条；

删除：删除光标所在的单元格；

清楚所有结果：清除运行笔记本后的结果；

折叠所有单元格：将当前笔记本内所有的单元格的代码和运行结果都折叠；

展开所有单元格：将当前笔记本内所有折叠的单元格的代码和运行结果都展开；

显示快捷帮助：您可以通过快捷键操作笔记本，如切换到编辑模式、运行单元格等，点击显示快捷帮助按钮将提示所有可用的快捷操作。

您可以在笔记本内创建多个单元格，在不同单元格中写不同的语句以完成各项操作。

添加新的 Cell

鼠标悬浮在单元格左侧，点击出现的 ➕ 按钮，您可以快速添加一个单元格，将添加在光标所在单元格的下方，您也可以长按 + 上方按钮上下拖动单元格以调整所有单元格的顺序。

运行

鼠标悬浮在单元格上将在右上角显示运行按钮，点击后将运行单元格内的代码，运行后您可以查询结果、任务详情或日志信息，如下图所示：

运行至此

当您对当前的 cell 执行了编辑，需要重新执行脚本直到当前 cell 停止时，可点击运行至此按钮，如下图所示：

下载

您也可以下载运行结果到 CSV 文件，如下图所示：

2.3.6 工作流 Workflow

Byzer Notebook 提供了可视化 Workflow 功能。您可以通过拖拽实现对数据集的导入、筛选、训练、预测和存储。

同时 Byzer Notebook 还提供了一键转码的功能。编辑好的 Workflow 可以通过右上角的开关一键转换为 Notebook，方便您的二次开发。

创建工作流

点击左侧导航栏工作区旁的 -> 工作流创建工作流文件。

输入工作流名称，并点击创建，您会自动进入到工作流页面

在该页面，您可以创建、编辑、重命名、克隆和删除工作流。

您也可以在左侧导航栏中的工作区模块对创建的工作流进行重命名、克隆、移动、导出和删除操作。

工作流节点

目前 Byzer 支持的工作流节点有五类：

1. Basic Node 基本节点：

Load：加载数据
Select：选择
Save：保存
Train：训练
Predict：预测
Register：注册

2. Algorithm Node 算法节点：

ALSInPlace：交替最小二乘法（Alternating Least Square），一种协同推荐算法。
TfldfinPlace：词频-逆向文件频率（Term Frequency-Inverse Document Frequency），是一种统计方法，用以评估一字词对于一个文件集或一个语料库中的其中一份文件的重要程度。
KMeans：K均值聚类算法。
LogisticRegression：逻辑回归是一个广义线性回归分析模型，常被用于二分类或多分类场景。
NaiveBayes：朴素贝叶斯，用于监督学习。
RandomForest：随机森林算法，利用多个决策树去训练，分类和预测样本。
Linear Regression：线性回归是一种统计分析方法，它利用回归分析来确定两种或两种以上变量间相互依赖的定量关系。

3. Feature Engineering Node 特征工程节点：

Discretizer：将连续的特征转化为离散的特征。
NormalizelnPlace：特征归一化，本质上是为了统一量纲，让一个向量里的元素变得可以比较。
ScalerlnPlace：特征平滑，可以将输入的特征中的异常数据平滑到一定范围。
Word2VeclnPlace：将文本内容转变成向量。

4. Data Processing Node 数据处理节点：

JsonExpandExt：可以轻松地将一个 JSON字段扩展为多个字段。
RateSampler：数据集切分，支持对每个分类的数据按比例切分。
TableRepartition：可以修改分区数量。例如，在我们保存文件前或使用 Python 时，我们需要使用 TableRepartition 来使 Python workers 尽可能地并行运行。

5. Tool Node 应用工具节点：

SyntaxAnalyzeExt: 被用来完成表格抽取和解析所有在 SQL 中的表。
TreeBuildExt: 被用来处理树状分析。

创建/删除节点

拖动您想要创建的节点至工作流画布页面任意位置，在弹窗中填写需要配置的参数，点击确定完成节点创建。您可以创建多个节点，创建完成后，Byzer Notebook 会根据节点之间的输入输出关系自动连线。您也可以在画布上拖动节点改变节点位置。

若您需要更新某个创建好的节点，点击该节点，在页面右侧的编辑框中根据您的需求修改节点参数即可。保存修改后，节点间的连线也会更新。

若您需要删除某个创建好的节点，点击该节点，再按 delete 或 backspace 即可删除该节点。

存储为笔记本

您可以通过两种方式将当前工作流存储为笔记本。

点击页面保存按钮，即可完成保存。
若您希望先行预览，点击画布右上角的笔记本页签，即可预览。

注意：这时您可以预览内容，但不可以对其进行编辑。

若内容确认无误，可点击页面保存按钮，并在弹窗中输入笔记本名称，再点击创建。创建成功后将自动进入该笔记本编辑页面。

注意：将工作流存储为笔记本后，原来的工作流依然存在，您可以在左侧导航栏的工作区模块中查看。

示例

在本例中，我们将对一个名为 feature 的文件进行切分，切分后的文件中的数据量分别为源文件的10%和90%。

创建一个新的工作流。
在左侧导航栏的数据目录模块中，点击 File System 旁的浮标 + ，并上传文件 features。

在左侧导航栏的工作流模块中，将 Basic Node 中的 Load 拖至工作流画布中，并填写相关配置参数：Data Source 选择 HDFS，Data Type 选择 json，Source File Name 选择 tmp -> upload -> features.txt，在 Output 中输入 result_1。最后点击确定，该节点创建成功。

在左侧导航栏的工作流模块中，将 Data Processing Node 中的 RateSampler 拖至工作流画布中，并填写相关配置参数：Select the Input Table 选择上一步中输出的 result_1，isSplitWithSubLable 默认为 true，labelCol 默认为 label，sampleRate 默认为 0.9,0.1，在 Output Name 中输入 rateSampler_result。最后点击确定，该节点创建成功。

创建的两个节点根据输入输出关系自动连线，如下图所示：

点击页面保存按钮，输入笔记本名称，将其保存为笔记本，页面将自动进入笔记本编辑页面。点击页面工具栏中的运行按钮，结果如下图所示：

2.4.开发流程说明

暂时无法在飞书文档外展示此内容

3.开发流程（重点学习）

3.1.权限申请

3.2.数据源配置

3.2.1 文本数据源

EXCEL:

load excel. ./example-data/excel/user-behavior.xlsx where header="true" as user_behavior; select cast(datatime as date) as day, sum(case when behavior_type = 'pv' then 1 else 0 end) as pv, count(distinct user_id) as uv from user_behavior group by cast(datatime as date) order by day as day_pv_uv;

CSV:

load csv.`tmp/upload/taxi_tour_table_train_simple.csv` where delimiter="," and header = "true" as taxi_tour_table_train_simple;

3.2.2 database数据源

参数说明：

user和password为key码
Holo 的endpoint 域名解析，在开发环境，需要单独在服务器/etc/hosts进行配置；（IP 需要找齐云升）；生成环境不需要，会自动解析
sl_data_prod为holo对应的项目空间名称
db_2为新命名的数据库名

SET user="XXXX"; SET password="XXXXX"; CONNECT jdbc WHERE url="jdbc:postgresql://holo-cn-fb4t8dpbb942-cn-baoding-gwmcloud-d01-internal.hologres.ops.cloud.gwm.cn:80/sl_data_prod" and driver="org.postgresql.Driver" and user="${user}" and password="${password}" AS db_2;

3.3.数据装载

3.3.1 文本数据的加载

先来看一个最简单的 load 语句：

set abc=''' { "x": 100, "y": 200, "z": 200 ,"dataType":"A group"} { "x": 120, "y": 100, "z": 260 ,"dataType":"B group"} '''; load jsonStr.`abc` as table1; select * from table1 as output;

load，代表读入数据源的行为
jsonStr ，这里代表的是数据源或者格式名称，该句表示这里加载的是一个 json 的字符串，
. 和 abc ，通常反引号内是个路径，比如:csv.`/tmp/csvfile`
为了方便引用加载的结果表，我们需要使用 as 句式，将结果存为一张新表，这里我们取名为 table1。table1 可以被后续的 select 句式引用。

3.3.2 database数据的加载

sl_data为sl_data_prod的实例名称，其他项目空间填写对应的实例名
加载后的数据表需重命名

load jdbc.`db_2.sl_data` where directQuery=''' SELECT * FROM t193_bc_tag_userprofile_attribute_a_d limit 10 ''' as newtable; select * from newtable as t_protecting_customer;

3.4.数据探索

3.4.1 数据可视化

YAML 格式详解：

YAML中的顶级元数有三个： 1. runtime 配置运行时。 YAML 文件会被转化为 Python 代码执行，所以runtime 其实是配置 Python环境。 2. control 控制图表的一些生成行为，比如是生成html还是image，数据是不是再需要一次排序等等 3. fig 描绘生成什么样的图表，该图表的配置是什么

runtime：

runtime 下只有一层子元数，常见配置如下。

env 指定需要使用的服务器端Python环境，如果发现运行错误，提示no module xxx，需要在对应的服务器端python虚拟环境进行pip install。
cache 图表结果是不是要缓存，如果你在其他cell要引用这个图标结果，需要设置为true。默认设置为false 即可。
output 将图表转化为一个表引用，方便后续 SQL 使用。默认可以不用配置。
runIn 在哪个类型节点执行。 driver/executor 。推荐 driver。

control：

ignoreSort 默认为true. 系统会对 X 轴字段进行默认进行排序 2. format 默认为 html。如果需要生成图片，可以设置为 image

fig:

fig.xxx 其中 xxx 为图标类型。支持 line,bar,pie等
fig.xxx.title 图表标题
fig.xxx.x X 轴。支持字符串或者数组配置
fig.xxx.y Y 轴。支持字符串或者数组配置
fig.xxx.labels 改动图标中的一些名称。默认为字典

一个较为完整的配置如下：

runtime: env: source /opt/miniconda3/bin/activate ray-1.12.0 cache: false output: jack control: ignoreSort: false format: image fig: bar: title: "日PV/UV柱状图" x: day y: - pv - uv labels: day: "日期"

示例：

load excel.`./example-data/excel/user-behavior.xlsx` where header="true" as user_behavior;select cast(datatime as date) as day,sum(case when behavior_type = 'pv' then 1 else 0 end) as pv,count(distinct user_id) as uvfrom user_behaviorgroup by cast(datatime as date)order by day as day_pv_uv;!visualize day_pv_uv ''' runtime: env: source /opt/miniconda3/bin/activate ray-1.12.0 cache: false output: jack control: ignoreSort: false format: image fig: bar: title: "日PV/UV柱状图" x: day y: - pv - uv labels: day: "日期" '''; select unbase64(content) as content, "wow.png" as fileName from jack as imageTable;save overwrite imageTable as image.`/tmp/images` where imageColumn="content" and fileName="fileName";-- !fs -ls /tmp/images; save overwrite command as Rest.`YOUR_UPLOAD_URL` where `config.method`="post"and `header.content-type`="multipart/form-data"345and `form.file-path`="/tmp/images/wow.png"and `form.file-name`="wow.png";

示例2：

#%visualize #%input=dash_data runtime: pythonExec: C:\Users\祝海林\.conda\envs\dev\python cache: false control: ignoreSort: True fig: bar: x: Deaths y: Entity animation_frame: Year orientation: h range_x: vv_type: code vv_value: "[0, df.Deaths.max()]" color: Entity title: "Evolution of Natural Disasters" labels: Entity: "自然灾害"

图可以自动播放

3.4.2 数据剖析

即描述性分析，包含：

列名 columnName
数据类型 dataType
唯一值比例 uniqueValueRatio
空（空值的比例）nullValueRatio
空白（空字符串的比例）blankValueRatio
均值 mean
中位数 (仅数值类型, 非数值类型展示为 0.0) median
众数 (仅数值类型，多个众数只显示一个，非数值类型展示 0) mode
标准差 (仅数值类型非数值类型展示为0)，standardDeviation
标准误差（见下方，仅数值类型 standardError
最小值 (仅数值类型) max
最大值 (仅数值类型) min
最小长度（该列中数据的最小长度）maximumLength
最大长度（该列中数据的最小长度）minimumLength
序号位置（字段在table中的位置/顺序）ordinalPosition
主健候选者（是/否，唯一值的比例为100%的字段，则“是”。）primaryKeyCandidate
非空计数（不是空值的数据量）nonNullCount
四分位数 -- %25
四分三位数 -- %75

-- 假设存在源表数据 table1 select * from table1 as table2; -- 执行 DataSummary 完成源表数据的数据剖析功能 run table2 as DataSummary.`` as summaryTable; -- 由于 v2.0 的 DataSummary 展示的剖析指标相对较多，用户可以根据需要选择指标，比如 run table2 as DataSummary.`` as summaryTable where metrics='mean, median, 75%' -- 选取剖析结果的均值，数据类型，众数，中位数还有3/4分位数作为指标输出 and roundAt='2' -- 剖析数据保留 2 位小数 and approxSwitch="false"; -- 是否精确计算分位数

3.4.3 模式识别

模式分布 ET 是字符串的模式分布统计的算法 ET。该 ET 主要是对标 SAS 系统的模式统计功能，总结字符串类型列的文本模式，从统计学的角度观察数据的模式分布，从而更好的结合数据离散化的加工。

set abc=''' {"name": "elena", "age": 57, "phone": 15552231521, "income": 433000, "label": 0} {"name": "candy", "age": 67, "phone": 15552231521, "income": 1200, "label": 0} {"name": "bob", "age": 57, "phone": 15252211521, "income": 89000, "label": 0} {"name": "candy", "age": 25, "phone": 15552211522, "income": 36000, "label": 1} {"name": "candy", "age": 31, "phone": 15552211521, "income": 300000, "label": 1} {"name": "finn", "age": 23, "phone": 15552211521, "income": 238000, "label": 1} '''; load jsonStr.`abc` as table1; select name, age, income from table1 as table2; run table2 as PatternDistribution.`` as pd_table; run table2 as PatternDistribution.`` where limit=1000 as pd_table;--limit 设置最多的模式行，如果模式总数不超过patternLimit的值，默认为100 run table2 as PatternDistribution.`` where excludeEmptyVal="true" as pd_table;--设置是否过滤空值，ture 或者 false，默认 true run table2 as PatternDistribution.`` where patternLimit=1000 as pd_table;--patternLimit 指- 设置最长的模式长度，默认为 1000

3.4.4 频数分布

帮助用户从统计的角度计算 count，输入是一张多列二维表，输出是一个两列的二维表，列1为字段名，列2为该字段的值的分布情况，并以 Json String 的方式展示

-- 假设存在源表数据 table1 select * from table1 as table2; -- 通过run/train关键字执行，频数分布为精确计算，内部有使用Action算子，会触发spark job提交 run table1 as DescriptiveMetrics.`` as descriptiveMetrics where metricSize='1000'; -- 支持通过参数metricSize控制条数，默认为100条。metricSize小于等于0是会报错提示参数设置错误。

返回结果示例：

columnName	descriptiveMetrics
age	[{"18":1},{"21":7}]
address	[{"上海":1},{"广州":7}]

3.4.5.唯一标识

在某些 EDA 需求中，需要对表生成一列全局唯一的值，该唯一值列为数字顺序递增，不会出现数据乱序，可以选择替换现有列或者创建新列

User Tutorial

通过Byzer ET的方式执行唯一标识符计算，可以设置可选参数来控制替换现有列或者创建新列，如果是创建新列，需要指定一个列名，默认为Unique_ID，新的列会插入到表的第一列前面；如果选择替换现有列，则会在原有列的位置，进行数据覆盖。

唯一值生成规则：唯一值为从 1 开始的自增ID，步长为 1

调用方式如下:

-- 假设存在源表数据 table1select * from table1 as table2;-- 调用唯一标识符计算的ET，返回值包括原始列名和频数分布的json数据 run table2 as UniqueIdentifier.`` where source="replace" and columnName="income" as uniqueIdentifier;

返回结果示例：

Unique_ID	havana_id	a	b
1	1	aaa	bbb
2	2	aaa	bbb
3	3	aaa	bbb
4	1	aaa	bbb
5	2	aaa	bbb
6	3	aaa	bbb

可选参数

source 设置替换现有列或者创建新列，new 或 replace，默认值 new。
columnName 设置唯一值列的列名，默认为 Unique_ID，如果输入的列存在，会报错终止操作。

3.5.数据转换

select 句式是 Byzer-lang 中处理数据最重要的方式之一。

Byzer-lang 中的 select 句式除了最后 as 表名 以外，完全兼容 Spark SQL。一般来讲，可以结合使用 Spark SQL 中的函数和算子以及 Byzer 的一些特定语法命令或 UDF 来完成数据转换的功能

3.5.1. 基本语法

最简单的一个 select 语句：

select 1 as col1 as table1;

结果为：col1: 1。

从上面代码可以看到，Byzer-lang 中的 select 语法和传统 SQL select 语法唯一的差别就是后面多了一个 as tableName。这也是为了方便后续对该 SQL 处理的结果进行引用引入的微小改良。

正常的 SQL 语句：

SELECT b.* FROM table_a as aLEFT JOIN table_b as b ON a.id = b.id WHERE a.study_id in( '12345678' )AND a.status <> 3 AND b.use_status = 0；

Byzer 语法：

SELECT b.* FROM table_a as aLEFT JOIN table_b as b ON a.id = b.id WHERE a.study_id in( '12345678' )AND a.status <> 3 AND b.use_status = 0 as new_table;select * from new_table as traindata;

比如，对于 new_table, 用户可以在新的 select 语句中进行引用：

3.5.2 Select 句式中的模板功能

实际在书写 select 语句可能会非常冗长。Byzer-lang 提供了两种方法帮助大家简化代码。

对于如下代码示例：

select "" as features, 1 as label as mockData;select SUM( case when features is null or features='' then 1 else 0 end ) as features,SUM( case when label is null or label='' then 1 else 0 end ) as label,1 as a from mockData as output;

如果字段特别多，而且都要做类似的事情，可能要写非常多的 SUM 语句。

用户可以通过如下语法进行改进：

select "" as features, 1 as label as mockData;select #set($colums=["features","label"])#foreach( $column in $colums )SUM( case when `$column` is null or `$column`='' then 1 else 0 end ) as $column,#end1 as a from mockData as output;

#set 设置了一个模板变量 $columns, 然后使用 #foreach 对该变量进行循环，里面的 SUM 本质上成了一个模板。系统在执行该 select 语句的时候，会自动根据这些指令展开成类似前面手写的代码。

Byzer-lang 还提供了一个更加易用的模板方案：

set sum_tpl = ''' SUM( case when {0} is null or {0} ='' then 1 else 0 end ) as {0} ''';select ${template.get("sum_tpl","label")}, ${template.get("sum_tpl","label")} from mockData as output;

通过变量声明设置一个模板，该模板通过名为 sum_tpl 变量持有，并且支持位置参数。接着，在 select 句式中使用 ${template.get} 对模板进行渲染了。第一个参数是模板名，后面的参数则是模板的参数。

3.6 特征工程

3.6.1 特征平滑

特征平滑算子 ScalerInPlace 可以将输入特征中包含异常的数据平滑到一定区间，支持阈值平滑和指数平滑等方式。ScalerInPlace 支持 min-max， log2，logn 等方法对数据进行特征平滑。下面将介绍该算子的应用方式。

1.数据准备

-- 创建数据集set jsonStr=''' {"a":1, "b":100, "label":0.0}, {"a":100, "b":100, "label":1.0} {"a":1000, "b":100, "label":0.0} {"a":10, "b":100, "label":0.0} {"a":1, "b":100, "label":1.0} ''';load jsonStr.`jsonStr` as data;

2.对数据进行特征平滑处理

-- 用 ScalerInPlace 对 a,b 两列数据都进行平滑: train data1 as ScalerInPlace.`/tmp/scaler`where inputCols="a,b"and scaleMethod="min-max"and removeOutlierValue="false";-- 将处理结果 load 出来查看load parquet.`/tmp/scaler/data` as featurize_table;

代码含义：

采用 ScalerInPlace 算子对数据集 data 进行训练处理，并将该组设置以特征工程模型的形式保存在/tmp/scaler 路径下。
参数 inputCols 用于设置参与处理的列名。
参数 scaleMethod 用于设置特征平滑的处理方式，可选项还有：log2，logn 。
参数 removeOutlierValue 控制是否自动填充异常值。若设置为true，则会自动用中位数填充异常值。

可以使用命令：!show "et/params/ScalerInPlace"; 查看该算子包含的所有参数的使用方式

结果如下：

3. API 预测

通过上面的命令，就可以将该特征平滑模型训练结果通过 register 语句注册成一个函数，这里命名为：scale_convert，注册成功的函数会把训练阶段学习到的东西应用起来。

现在，任意给定两个数字，都可以使用 scale_convert 函数将其平滑：

select scale_convert(array(cast(7.0 as double), cast(8.0 as double))) as features as output;

输出结果为：

features [0.006006006006006006,0.5]

3.6.2 归一化

特征归一化本质上是为了统一量纲，让一个向量里的元素变得可以比较。

它应用于任何依赖于距离的算法，比如 KMeans，Nearest Neighbors Methods， RBF Kernels 等等。

1. 数据准备

-- 创建数据集，这里创建一份js string 类型的set jsonStr=''' {"a":1, "b":100, "label":0.0}, {"a":100, "b":100, "label":1.0} {"a":1000, "b":100, "label":0.0} {"a":10, "b":100, "label":0.0} {"a":1, "b":100, "label":1.0} ''';load jsonStr.`jsonStr` as data;

2. 归一化操作

train data as NormalizeInPlace.`/tmp/model`where inputCols="a,b"and scaleMethod="standard"and removeOutlierValue="false";load parquet.`/tmp/model/data` as output;

代码含义：

采用 NormalizeInPlace 算子对数据集 data 进行训练处理，并将该组设置以特征工程模型的形式保存在/tmp/model 路径下。
参数 inputCols 用于设置参与处理的列名。
参数 scaleMethod 用于设置归一化的处理方式，可选项还有：p-norm 。
参数 removeOutlierValue 控制是否自动填充异常值。若设置为true，则会自动用中位数填充异常值。

如果 inputCols 只有一列，那么该列可以为 double 数组

可以使用命令：!show "et/params/NormalizeInPlace"; 查看该算子包含的所有参数的使用方式

结果如下：

3. API 预测

通过上面的命令，就可以将该归一化模型训练结果通过 register 语句注册成一个函数，这里命名为：convert，注册成功的函数会把训练阶段学习到的东西应用起来。

现在，任意给定两个数字，都可以使用 convert 函数将内容归一化处理。

select convert(array(cast(7.0 as double), cast(8.0 as double))) as features as output;

输出结果为：

features [ -0.4932558994483363, 0 ]

3.6.3 混淆矩阵

混淆矩阵其实就是把所有类别的预测结果与真实结果按类别放置到了同一矩阵中，在这个矩阵中我们可以清楚看到每个类别正确识别的数量和错误识别的数量。在分类算法里用处很多，用户可以直观看到数据的错误分布情况。

1. 数据准备

假设我们有动物分类，两列中一列是实际值，另一列是预测值，内容如下：

set rawData=''' {"label":"cat","predict":"rabbit"} {"label":"cat","predict":"dog"} {"label":"cat","predict":"cat"} {"label":"dog","predict":"dog"} {"label":"cat","predict":"dog"} ''';load jsonStr.`rawData` as data;

2. 训练

使用混淆矩阵来查看最后的预测结果分布：

train data as ConfusionMatrix.`/tmp/model` where actualCol="label" and predictCol="predict";load parquet.`/tmp/model/data` as output;

代码含义：

采用 ConfusionMatrix 算子对数据集 data 进行训练处理，并将该组设置以特征工程模型的形式保存在/tmp/model 路径下。
参数 actualCol 用于指定实际值所在列名。
参数 predictCol 用于指定预测值所在列名。

结果如下：

另外我们也可以看到一些统计值：

load parquet.`/tmp/model/detail` as output;

结果如下：

3.6.4 数据切分

在做算法时，我们需要经常对数据切分成训练集和测试集。 RateSampler 算子支持对数据集进行按比例切分。

1. 数据准备

-- 创建数据集set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;

2. 切分数据集

现在我们使用 RateSampler 进行切分：

train data as RateSampler.`/tmp/model` where labelCol="label"and sampleRate="0.7,0.3" as marked_dataset;select * from marked_dataset as output;

代码含义：

采用 RateSampler 算子对数据集 data 进行训练处理，并将该组设置以特征工程模型的形式保存在/tmp/model 路径下。
参数 labelCol 用于指定切分的字段。
参数 sampleRate 用于指定切分的比例。

可以使用命令：!show "et/params/RateSampler"; 查看该算子包含的所有参数的使用方式

结果如下：

数据集多出了一个字段 split, 0 表示前一个集合（训练集）， 1 表示后一个集合（测试集）。

接着，可以这样切分出训练集和测试集：

-- 筛选出训练集select * from marked_dataset where __split__=0as trainingTable;-- 筛选出测试集select * from marked_dataset where __split__=1as validateTable;

默认 RateSampler 采用估算算法。如果数据集较小，可以通过设置参数 isSplitWithSubLabel="true" 获得非常精确的划分。

3.7.向量化

label为y值，features为特征值，数据需要为dobule格式

select vec_dense(array( passenger_count_d, vendor_sum_pl, vendor_max_pl, vendor_min_pl, vendor_avg_pl, pc_sum_pl, pc_max_pl, pc_min_pl, pc_avg_pl, pc_cnt_d, vendor_cnt )) as features,cast(trip_duration as double) as label from new_feature_data as training_table;

3.8.机器学习

3.8.1 Kmeans

KMeans，k均值聚类算法（k-means clustering algorithm）是一种迭代求解的聚类分析算法，其步骤是，预将数据分为K组，则随机选取K个对象作为初始的聚类中心，然后计算每个对象与各个种子聚类中心之间的距离，把每个对象分配给距离它最近的聚类中心。

数据准备：

结果如下：

聚类算法属于无监督算法，所以没有 Label 的概念。接着，我们可以训练了：

train data1 as KMeans.`/tmp/alg/kmeans`where k="2"and seed="1";

API 预测

训练完成后，可以注册模型为函数，进行预测：

结果如下：

3.8.2.NaiveBayes

NaiveBayes 是一种分类算法。和决策树模型相比，朴素贝叶斯分类器(Naive Bayes Classifier 或 NBC)发源于古典数学理论，有着坚实的数学基础，以及稳定的分类效率。同时，NBC模型所需估计的参数很少，对缺失数据不太敏感，算法也比较简单。

-- 创建测试数据set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;select vec_dense(features) as features ,label as label from dataas data1;-- 用朴素贝叶斯训练数据 train data1 as NaiveBayes.`/tmp/model` where-- 如果参数 keepVersion 设置成 true，以后每次运行脚本，Byzer 都会为你的模型保存一个最新的版本 keepVersion="true" -- 用参数 evaluateTable 指明验证集，它将被用来给评估器提供一些评价指标，如：F1、准确度等and evaluateTable="data1"-- 指明参数组0（即：第一组参数组）的参数and `fitParam.0.featuresCol`="features"and `fitParam.0.labelCol`="label"and `fitParam.0.smoothing`="0.5"-- 指明参数组1（即：第二组参数组）的参数and `fitParam.1.featuresCol`="features"and `fitParam.1.labelCol`="label"and `fitParam.1.smoothing`="0.2";

最后输出结果如下：

name value --------------------------------- modelPath /tmp/model/_model_10/model/1 algIndex 1 alg org.apache.spark.ml.classification.NaiveBayes metrics f1: 0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success startTime 20180913 59:15:32:685 endTime 20180913 59:15:36:317 trainParams Map(smoothing -> 0.2,featuresCol -> features, labelCol -> label) --------------------------------- modelPath /tmp/model/_model_10/model/0 algIndex 0 alg org.apache.spark.ml.classification.NaiveBayes metrics f1:0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success startTime 20180913 59:1536:318 endTime 20180913 59:1538:024 trainParams Map(smoothing -> 0.2, featuresCol -> features, labelCol -> label)

对于大部分内置算法而言，都支持如下几个特性：

可以通过 keepVersion 来设置是否保留版本。
通过 fitParam. 数字序号 配置多组参数，设置 evaluateTable 后系统自动算出 metrics.

批量预测

predict data1 as NaiveBayes./tmp/model;

结果如下：

features label rawPrediction probability prediction {"type":1,"values":[5.1,3.5,1.4,0.2]} 0 {"type":1,"values":[16.28594461094461,3.7140553890553893]} {"type":1,"values":[0.8142972305472306,0.18570276945276948]} 0 {"type":1,"values":[5.1,3.5,1.4,0.2]} 1 {"type":1,"values":[16.28594461094461,3.7140553890553893]} {"type":1,"values":[0.8142972305472306,0.18570276945276948]} 0

API 预测

register NaiveBayes.`/tmp/model` as rf_predict;-- 参数 algIndex 你可以指明用哪一组参数训练出的模型 register NaiveBayes.`/tmp/model` as rf_predict where algIndex="0";-- 参数 autoSelectByMetric 可以用来指明用那个指标来判断最优模型 register NaiveBayes.`/tmp/model` as rf_predict where autoSelectByMetric="f1";select rf_predict(features) as predict_label, label from data1 as output;

参数algIndex 可以让用户指定用哪组参数得到算法模型。
当然用户也可以让系统自动选择，前提是在训练时配置了参数evalateTable 预先评估模型的表现情况，然后使用参数 autoSelectByMetric 指定判定指标即可选出最优算法模型。
最后，就可以像使用一个函数一样对一个 feature 进行预测了。

3.8.3 RandomForest

RandomForest 随机森林是利用多个决策树对样本进行训练、分类并预测的一种分类算法，主要应用于回归和分类场景。在对数据进行分类的同时，还可以给出各个变量的重要性评分，评估各个变量在分类中所起的作用。

-- 创建测试数据集set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;select vec_dense(features) as features ,label as label from dataas data1;-- 使用随机森林算法进行训练 train data1 as RandomForest.`/tmp/model` where-- 如果参数 keepVersion 设置成 true，以后每次运行脚本，Byzer 都会为你的模型保存一个最新的版本 keepVersion="true" -- 用参数 evaluateTable 指明验证集，它将被用来给评估器提供一些评价指标，如：F1、准确度等and evaluateTable="data1"-- 指明参数组0（即：第一组参数组）的参数and `fitParam.0.featuresCol`="features"and `fitParam.0.labelCol`="label"and `fitParam.0.maxDepth`="2"-- 指明参数组1（即：第二组参数组）的参数and `fitParam.1.featuresCol`="features"and `fitParam.1.labelCol`="label"and `fitParam.1.maxDepth`="10";

最后输出结果如下：

name value --------------------------------- modelPath /tmp/model/_model_10/model/1 algIndex 1 alg org.apache.spark.ml.classification.RandomForestClassifier metrics f1: 0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success startTime 20180913 59:15:32:685 endTime 20180913 59:15:36:317 trainParams Map(maxDepth -> 10) --------------------------------- modelPath /tmp/model/_model_10/model/0 algIndex 0 alg org.apache.spark.ml.classification.RandomForestClassifier metrics f1:0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success startTime 20180913 59:1536:318 endTime 20180913 59:1538:024 trainParams Map(maxDepth -> 2, featuresCol -> features, labelCol -> label)

对于大部分内置算法而言，都支持如下几个特性：

可以通过 keepVersion 来设置是否保留版本。
通过 fitParam.数字序号 配置多组参数，设置 evaluateTable 后系统自动算出 metrics.

批量预测

predict data1 as RandomForest./tmp/model;

结果如下：

API预测

register RandomForest.`/tmp/model` as rf_predict;-- 参数 algIndex 你可以指明用哪一组参数训练出的模型 register RandomForest.`/tmp/model` as rf_predict where algIndex="0";-- 参数 autoSelectByMetric 可以用来指明用那个指标来判断最优模型,此处选择 F1 register RandomForest.`/tmp/model` as rf_predict where autoSelectByMetric="f1";select rf_predict(features) as predict_label, label from data1 as output;

参数algIndex 可以让用户指定用哪组参数得到算法模型。
当然用户也可以让系统自动选择，前提是在训练时配置了参数evalateTable 预先评估模型的表现情况，然后使用参数 autoSelectByMetric 指定判定指标即可选出最优算法模型。
最后，就可以像使用一个函数一样对一个 feature 进行预测了。
Usage tips

##### Usage tips： We include a few guidelines for using random forests by discussing the various parameters. We omit some decision tree parameters since those are covered in the decision tree guide. The first two parameters we mention are the most important, and tuning them can often improve performance: numTrees: Number of trees in the forest. Increasing the number of trees will decrease the variance in predictions, improving the model’s test-time accuracy. Training time increases roughly linearly in the number of trees. maxDepth: Maximum depth of each tree in the forest. Increasing the depth makes the model more expressive and powerful. However, deep trees take longer to train and are also more prone to overfitting. In general, it is acceptable to train deeper trees when using random forests than when using a single decision tree. One tree is more likely to overfit than a random forest (because of the variance reduction from averaging multiple trees in the forest). The next two parameters generally do not require tuning. However, they can be tuned to speed up training. subsamplingRate: This parameter specifies the size of the dataset used for training each tree in the forest, as a fraction of the size of the original dataset. The default (1.0) is recommended, but decreasing this fraction can speed up training. featureSubsetStrategy: Number of features to use as candidates for splitting at each tree node. The number is specified as a fraction or function of the total number of features. Decreasing this number will speed up training, but can sometimes impact performance if too low. 参考： https://spark.apache.org/docs/3.5.1/mllib-ensembles.html

3.8.4 LogisticsRegression

Logistic Regression 一种广义的线性回归分析模型，常用于数据挖掘，疾病自动诊断，经济预测等领域。

-- 创建测试数据set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;select vec_dense(features) as features , label as label from dataas data1;-- select * from data1 as output1;-- 使用逻辑回归 train data1 as LogisticRegression.`/tmp/model_2` where-- 如果参数 keepVersion 设置成 true，以后每次运行脚本，Byzer 都会为你的模型保存一个最新的版本 keepVersion="true" -- 用参数 evaluateTable 指明验证集，它将被用来给评估器提供一些评价指标，如：F1、准确度等and evaluateTable="data1"-- 指明参数组0（即：第一组参数组）的参数and `fitParam.0.labelCol`="label"and `fitParam.0.featuresCol`="features"and `fitParam.0.fitIntercept`="true"-- 指明参数组1（即：第二组参数组）的参数and `fitParam.1.featuresCol`="features"and `fitParam.1.labelCol`="label"and `fitParam.1.fitIntercept`="false";

最后输出结果如下：

name value --------------- ------------------ modelPath /_model_5/model/1 algIndex 1 alg org.apache.spark.ml.classification.LogisticRegression metrics f1: 0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success message startTime 20210824 42:14:33:761 endTime 20210824 42:14:41:984 trainParams Map(labelCol -> label, featuresCol -> features, fitIntercept -> false) --------------- ------------------ modelPath /_model_5/model/0 algIndex 0 alg org.apache.spark.ml.classification.LogisticRegression metrics f1: 0.7625000000000001 weightedPrecision: 0.8444444444444446 weightedRecall: 0.7999999999999999 accuracy: 0.8 status success message startTime 20210824 42:14:41:985 endTime 20210824 42:14:47:830 trainParams Map(featuresCol -> features, labelCol -> label, fitIntercept -> true)

对于大部分内置算法而言，都支持如下几个特性：

可以通过 keepVersion 来设置是否保留版本。
通过 fitParam.数字序号 配置多组参数，设置 evaluateTable 后系统自动算出 metrics.

批量预测

predict data1 as LogisticRegression./tmp/model;

结果如下：

features label rawPrediction probability prediction {"type":1,"values":[5.1,3.5,1.4,0.2]} 0 {"type":1,"values":[1.0986123051777668,-1.0986123051777668]} {"type":1,"values":[0.7500000030955607,0.24999999690443933]} 0 {"type":1,"values":[5.1,3.5,1.4,0.2]} 1 {"type":1,"values":[1.0986123051777668,-1.0986123051777668]} {"type":1,"values":[0.7500000030955607,0.24999999690443933]} 0

API预测

register LogisticRegression.`/tmp/model_2` as lr_predict;-- 参数 algIndex 你可以指明用哪一组参数训练出的模型 register LogisticRegression.`/tmp/model_2` as lr_predict where algIndex="0";-- 参数 autoSelectByMetric 可以用来指明用那个指标来判断最优模型,此处选择 F1 register LogisticRegression.`/tmp/model_2` as lr_predict where autoSelectByMetric="f1";select lr_predict(features) as predict_label, label from data1 as output;

参数algIndex 可以让用户指定用哪组参数得到算法模型。
当然用户也可以让系统自动选择，前提是在训练时配置了参数evalateTable 预先评估模型的表现情况，然后使用参数 autoSelectByMetric 指定判定指标即可选出最优算法模型。
最后，就可以像使用一个函数一样对一个 feature 进行预测了。

3.8.5 LinearRegression

线性回归 (Linear Regression) 是利用称为线性回归方程的最小二乘函数对一个或多个自变量和因变量之间关系进行建模的一种回归分析。

-- 创建测试数据set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;select vec_dense(features) as features , label as label from dataas data1;-- select * from data1 as output1;-- 使用线性回归 train data1 as LinearRegression.`/tmp/model_3` where-- 如果参数 keepVersion 设置成 true，以后每次运行脚本，Byzer 都会为你的模型保存一个最新的版本 keepVersion="true" -- 用参数 evaluateTable 指明验证集，它将被用来给评估器提供一些评价指标，如：F1、准确度等and evaluateTable="data1"-- 指明参数组0（即：第一组参数组）的参数and `fitParam.0.labelCol`="label"and `fitParam.0.featuresCol`="features"and `fitParam.0.elasticNetParam`="0.1"-- 指明参数组1（即：第二组参数组）的参数and `fitParam.1.featuresCol`="features"and `fitParam.1.labelCol`="label"and `fitParam.1.elasticNetParam`="0.8";

最后输出结果如下：

name value --------------- ------------------ modelPath /_model_4/model/1 algIndex 1 alg org.apache.spark.ml.regression.LinearRegression metrics f1: 0.0 weightedPrecision: 0.0 weightedRecall: 0.0 accuracy: 0.0 status success message startTime 20210824 52:14:52:441 endTime 20210824 52:14:53:429 trainParams Map(labelCol -> label, featuresCol -> features, elasticNetParam -> 0.8) --------------------------------- modelPath /_model_4/model/0 algIndex 0 alg org.apache.spark.ml.regression.LinearRegression metrics f1: 0.0 weightedPrecision: 0.0 weightedRecall: 0.0 accuracy: 0.0 status success message startTime 20210824 52:14:53:429 endTime 20210824 52:14:54:228 trainParams Map(featuresCol -> features, elasticNetParam -> 0.1, labelCol -> label)

对于大部分内置算法而言，都支持如下几个特性：

可以通过 keepVersion 来设置是否保留版本。
通过 fitParam.数字序号 配置多组参数，设置 evaluateTable 后系统自动算出 metrics.

批量预测

predict data1 as LinearRegression.`/tmp/model_3`;

结果如下：

features label prediction {"type":1,"values":[5.1,3.5,1.4,0.2]} 0 0.24999999999999645 {"type":1,"values":[5.1,3.5,1.4,0.2]} 1 0.24999999999999645

API预测

register LinearRegression.`/tmp/model_3` as lr_predict;-- 参数 algIndex 你可以指明用哪一组参数训练出的模型 register LinearRegression.`/tmp/model_3` as lr_predict where algIndex="0";-- 参数 autoSelectByMetric 可以用来指明用那个指标来判断最优模型,此处选择 F1 register LinearRegression.`/tmp/model_3` as lr_predict where autoSelectByMetric="f1";select lr_predict(features) as predict_label, label from data1 as output;

参数 algIndex 可以让用户指定用哪组参数得到算法模型。
当然用户也可以让系统自动选择，前提是在训练时配置了参数evalateTable 预先评估模型的表现情况，然后使用参数 autoSelectByMetric 指定判定指标即可选出最优算法模型。
最后，就可以像使用一个函数一样对一个 feature 进行预测了。

3.8.6 AutoML

AutoML 是将机器学习应用于现实问题的端到端流程自动化的过程。

AutoML 可以提供将分类算法进行遍历训练的功能，这些算法包含 NaiveBayes， LogisticRegression，LinearRegression， RandomForest 以及 GBT 分类算法。AutoML 插件会对用户的输入数据进行多模型训练，然后针对模型表现指标，进行模型排序，给用户返回表现最优的算法模型。

-- 创建测试数据set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} ''';load jsonStr.`jsonStr` as data;select vec_dense(features) as features ,label as label from dataas data1; train data1 as AutoML.`/tmp/auto_ml` where-- 如果参数 algos 不设置，数据就会自动被以下这些算法训练：GBTs,LinearRegression,LogisticRegression,NaiveBayes,RandomForest algos="LogisticRegression,NaiveBayes" -- 如果参数 keepVersion 设置成 true，以后每次运行脚本，Byzer 都会为你的模型保存一个最新的版本and keepVersion="true" -- 用参数 evaluateTable 指明验证集，它将被用来给评估器提供一些评价指标，如：F1、准确度等and evaluateTable="data1";

最后输出结果如下：

AutoML支持如下几个特性：

可以通过参数 keepVersion 来设置是否保留版本。
AutoML 支持在用户指定的算法集合里进行模型训练，用户通过配置 algos 参数（目前支持 " GBTs, LinearRegression, LogisticRegression, NaiveBayes, RandomForest " 的子集），让数据集在指定的算法集合中进行训练，获取最优模型
AutoML 会根据算法的表现排序，默认是按照 accuracy，用户可以指定按照 f1 或者其他的 metrics 进行排序。
AutoML 预测的时候，会根据历史训练的所有模型中挑选出表现最好的模型进行打分预测，用户无需指定特定模型。

批量预测

用户可以通过 predict 语法来完成对数据集的批量预测，以下 Byzer 代码的解释为：

用 predict 语法预测数据集 data1 通过被保存在路径/tmp/auto_ml 下的 AutoML 模型训练后得到的结果

predict data1 as AutoML.`/tmp/auto_ml`;

结果如下：

3.8.7.Gradient-Boosted Trees (GBTs)

Gradient-Boosted Trees (GBTs) are ensembles of decision trees. GBTs iteratively train decision trees in order to minimize a loss function. Like decision trees, GBTs handle categorical features, extend to the multiclass classification setting, do not require feature scaling, and are able to capture non-linearities and feature interactions.

spark.mllib supports GBTs for binary classification and for regression, using both continuous and categorical features. spark.mllib implements GBTs using the existing decision tree implementation. Please see the decision tree guide for more information on trees.

Note: GBTs do not yet support multiclass classification. For multiclass problems, please use decision trees or Random Forests.

Basic algorithm

Gradient boosting iteratively trains a sequence of decision trees. On each iteration, the algorithm uses the current ensemble to predict the label of each training instance and then compares the prediction with the true label. The dataset is re-labeled to put more emphasis on training instances with poor predictions. Thus, in the next iteration, the decision tree will help correct for previous mistakes.

The specific mechanism for re-labeling instances is defined by a loss function (discussed below). With each iteration, GBTs further reduce this loss function on the training data.

Losses

The table below lists the losses currently supported by GBTs in spark.mllib. Note that each loss is applicable to one of classification or regression, not both.

Notation: NN = number of instances. yiyi = label of instance ii. xixi = features of instance ii. F(xi)F(xi) = model’s predicted label for instance ii.

Loss	Task	Formula	Description
Log Loss	Classification	2∑Ni=1log(1+exp(−2yiF(xi)))2∑i=1Nlog⁡(1+exp⁡(−2yiF(xi)))	Twice binomial negative log likelihood.
Squared Error	Regression	∑Ni=1(yi−F(xi))2∑i=1N(yi−F(xi))2	Also called L2 loss. Default loss for regression tasks.
Absolute Error	Regression	∑Ni=1\|yi−F(xi)\|∑i=1N\|yi−F(xi)\|	Also called L1 loss. Can be more robust to outliers than Squared Error.

Usage tips

We include a few guidelines for using GBTs by discussing the various parameters. We omit some decision tree parameters since those are covered in the decision tree guide.

loss: See the section above for information on losses and their applicability to tasks (classification vs. regression). Different losses can give significantly different results, depending on the dataset.
numIterations: This sets the number of trees in the ensemble. Each iteration produces one tree. Increasing this number makes the model more expressive, improving training data accuracy. However, test-time accuracy may suffer if this is too large.
learningRate: This parameter should not need to be tuned. If the algorithm behavior seems unstable, decreasing this value may improve stability.
algo: The algorithm or task (classification vs. regression) is set using the tree [Strategy] parameter.

Validation while training

Gradient boosting can overfit when trained with more trees. In order to prevent overfitting, it is useful to validate while training. The method runWithValidation has been provided to make use of this option. It takes a pair of RDD’s as arguments, the first one being the training dataset and the second being the validation dataset.

The training is stopped when the improvement in the validation error is not more than a certain tolerance (supplied by the validationTol argument in BoostingStrategy). In practice, the validation error decreases initially and later increases. There might be cases in which the validation error does not change monotonically, and the user is advised to set a large enough negative tolerance and examine the validation curve using evaluateEachIteration (which gives the error or loss per iteration) to tune the number of iterations.

参考：https://spark.apache.org/docs/3.5.1/mllib-ensembles.html

3.9.模型注册

训练成功得到模型文件后，就可以将模型注册成 UDF 函数供后续将模型应用于批，流，API中。此时可以使用 register 句式完成模型到 UDF 的注册。

register RandomForest.`/tmp/models/randomforest` as model_predict;select vec_array(model_predict(features)) as predicted_value from mock_data as output;

3.10.模型推理

predict 顾名思义，应该和机器学习相关预测相关。比如上面的 train 示例中，用户将随机森林的模型放在了 /tmp/models/randomforest 目录下，用户可以通过 predict 语句加载该模型，并且对表 testData 进行预测。

示例代码如下：

predict testData as RandomForest.`/tmp/models/randomforest`;

3.11.部署上线

在 Byzer 中，我们可以使用和内置算法一样的方式将一个基于 Byzer-python 训练出的 AI 模型注册成一个 UDF 函数，这样可以将模型应用于批、流，以及 Web 服务中。接下来我们将展示 Byzer-python 基于 Ray 从模型训练再到模型部署的全流程 demo。

1. 数据准备

首先，安装 tensorflow 和 keras:

pip install keras tensorflow "tenacity~=6.2.0"

准备 mnist 数据集（需要）：

!python env "PYTHON_ENV=source activate ray1.8.0";!python conf "schema=st(field(image,array(long)),field(label,long),field(tag,string))";!python conf "runIn=driver";!python conf "dataMode=model"; run command as Ray.`` where inputTable="command"and outputTable="mnist_data"and code=''' from pyjava.api.mlsql import RayContext, PythonContext from keras.datasets import mnist ray_context = RayContext.connect(globals(), None) (x_train, y_train),(x_test, y_test) = mnist.load_data() train_images = x_train.reshape((x_train.shape[0], 28 * 28)) test_images = x_test.reshape((x_test.shape[0], 28 * 28)) train_data = [{"image": image.tolist(), "label": int(label), "tag": "train"} for (image, label) in zip(train_images, y_train)] test_data = [{"image": image.tolist(), "label": int(label), "tag": "test"} for (image, label) in zip(test_images, y_test)] context.build_result(train_data + test_data) ''';save overwrite mnist_data as delta.`ai_datasets.mnist`;

上面的 Byzer-python 脚本，获取keras自带的 mnist 数据集，再将数据集保存到数据湖中。

2. 训练模型

接着就开始拿测试数据 minist 进行训练，下面是模型训练代码：

-- 获取训练数据集load delta.`ai_datasets.mnist` as mnist_data;select image, label from mnist_data where tag="train" as mnist_train_data;!python env "PYTHON_ENV=source activate ray1.8.0";!python conf "schema=file";!python conf "dataMode=model";!python conf "runIn=driver"; run command as Ray.`` where inputTable="mnist_train_data"and outputTable="mnist_model"and code=''' import ray import os import tensorflow as tf from pyjava.api.mlsql import RayContext from pyjava.storage import streaming_tar import numpy as np ray_context = RayContext.connect(globals(),"127.0.0.1:10001") data_servers = ray_context.data_servers() train_dataset = [item for item in RayContext.collect_from(data_servers)] x_train = np.array([np.array(item["image"]) for item in train_dataset]) y_train = np.array([item["label"] for item in train_dataset]) x_train = x_train.reshape((len(x_train),28, 28)) model = tf.keras.models.Sequential([ tf.keras.layers.Flatten(input_shape=(28, 28)), tf.keras.layers.Dense(128, activation='relu'), tf.keras.layers.Dropout(0.2), tf.keras.layers.Dense(10, activation='softmax') ]) model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy']) model.fit(x_train, y_train, epochs=5) model_path = os.path.join("tmp","minist_model") model.save(model_path) model_binary = [item for item in streaming_tar.build_rows_from_file(model_path)] ray_context.build_result(model_binary) ''';

最后把模型保存至数据湖里：

save overwrite mnist_model as delta.`ai_model.mnist_model`;

3. 将模型注册成 UDF 函数

训练好模型之后，我们就可以用 Byzer-lang 的 Register 语法将模型注册成基于 Ray 的服务了，下面是模型注册的代码：

!python env "PYTHON_ENV=source activate ray1.8.0";!python conf "schema=st(field(content,string))";!python conf "mode=model";!python conf "runIn=driver";!python conf "rayAddress=127.0.0.1:10001";-- 加载前面训练好的tf模型load delta.`ai_model.mnist_model` as mnist_model;-- 把模型注册成udf函数 register Ray.`mnist_model` as model_predict where maxConcurrency="8"and debugMode="true"and registerCode=''' import ray import numpy as np from pyjava.api.mlsql import RayContext from pyjava.udf import UDFMaster,UDFWorker,UDFBuilder,UDFBuildInFunc ray_context = RayContext.connect(globals(), context.conf["rayAddress"]) # 预测函数 def predict_func(model,v): test_images = np.array([v]) predictions = model.predict(test_images.reshape((1,28*28))) return {"value":[[float(np.argmax(item)) for item in predictions]]} # 将预测函数提交到 ray_context UDFBuilder.build(ray_context,UDFBuildInFunc.init_tf,predict_func) ''' and predictCode=''' import ray from pyjava.api.mlsql import RayContext from pyjava.udf import UDFMaster,UDFWorker,UDFBuilder,UDFBuildInFunc ray_context = RayContext.connect(globals(), context.conf["rayAddress"]) # UDFBuilder.apply(ray_context) ''';

这里 UDFBuilder 与 UDFBuildInFunc 都是 Pyjava 提供的高阶 API，用来将 Python 脚本注册成 UDF 函数。

4. 使用模型做预测

Byzer 提供了类 SQL 语句做批量（Batch）查询，加载您的数据集，即可对数据进行预测。

load delta.`ai_datasets.mnist` as mnist_data;select cast(image as array<double>) as image, label as label from mnist_data where tag = "test" limit 100 as mnist_test_data;select model_predict(array(image))[0][0] as predicted, label as label from mnist_test_data as output;

后续可以直接调用 Byzer-engine 的 Rest API，使用注册好的 UDF 函数对您的数据集作预测。

3.12.UDF

register ScriptUDF.`` as arrayLast where lang="scala" and code='''def apply(a:Seq[String])={ a.last }''' and udfType="udf";

3.13.Branch statement

select 1 as a as mockTable; set b_count=`select count(*) from mockTable ` where type="sql" and mode="runtime"; !if ''':b_count == 1 '''; select 1 as a as final_table; !else; select 2 as a as final_table; !fi; select * from final_table as output;

3.14. Module

include lib.`gitee.com/allwefantasy/lib-core` where alias="libCore";

include local.`libCore.udf.hello`; select hello() as name as output;

3.15.Function

set loadExcel = ''' load excel.`{0}` where header="true" as {1} '''; !loadExcel ./example-data/excel/hello_world.xlsx helloTable;

set loadExcel = ''' load excel.`${path}` where header="true" as ${tableName} '''; !loadExcel _ -path ./example-data/excel/hello_world.xlsx -tableName helloTable;

3.16.扩展

目前这个方法不可用，报错。只做一下备注。未来解决错误。

!plugin app add - "byzer-shell-3.3";

3.17. Variable 变量

set v1 = "a" where scope="session";

select "${v1}" as col as output;

4.MLSQL 语法（语法参考）

4.1语言向导

Byzer-lang 拥有声明式语言的特点，这和 SQL 非常类似。不同的是，Byzer-lang 也支持 Python 脚本，用户也可以使用 Scala/Java 动态开发和注册 UDF 函数，这使得其灵活度得到了很大提高。

Byzer-lang 针对大数据领域的流程抽象出了如下几个句法结构：

数据加载/Load
数据转换/Select
数据保存/Save
代码引入/Include
宏函数/!Macro
变量设置/Set
分支语句/!If|!Else

而针对机器学习领域，也做了类似的抽象：

模型训练/Train｜Run
模型注册/Register
模型预测/Predict

此外，在代码复用上，Byzer-lang 支持脚本和包的管理。

4.2.数据加载/Load

load excel.`./example-data/excel/hello_world.xlsx` where header="true" as hello_world;

在上面的语句中，通过 load 关键字进行加载申明。加载的数据格式为 excel , 路径为 ./example-data/excel/hello_world.xlsx, 加载的过程中配置参数在 where/options 子语句中。加载后的结果是一张表，使用 as 语法 进行命名，名称为 hello_world。

Byzer-lang 几乎可以加载市面上主流的数据源和数据格式：

数据源： JDBC 协议的数据库，多种云上对象存储，HDFS...
数据格式：例如 text， image， csv， json， xml 等文件格式

4.3.数据转换/Select

Byzer-lang 可以使用 select 句式处理加载后的数据。

load excel.`./example-data/excel/hello_world.xlsx` where header="true" as hello_world;select hello from hello_world as output;

在这里例子中使用兼容 Spark SQL 的语法对 hello-world 表进行处理。

4.4.数据保存/Save

load excel.`./example-data/excel/hello_world.xlsx` where header="true" as hello_world;save overwrite hello_world as csv.`./tmp/hw` where header="true";

save 表示要对表进行保存。overwrite 为两种 mode 中（overwrite，append）的一种，表示对目标对象进行覆盖。 hello_world 则是被保存的表。 as 后面紧接着保存的格式和路径。最后保存的配置选项在 where/options 子句中。

4.5.代码引入/Include

在同一个项目里，脚本之间的引用可以通过 include 句式来完成。

include project.`./src/common/PyHeader.mlsql`;

这个语法在桌面版中有效。不同的 Notebook 实现，则可能会有不同，但总体格式是一致的。

如果希望达到包级别的复用，Byzer-lang 使用 Github 作为包管理器。举例，lib-core 是 Byzer-lang 的一个示例 Lib：

https://github.com/allwefantasy/lib-core

可以通过如下代码将该库引入自己的项目：

include lib.`github.com/allwefantasy/lib-core`where libMirror="gitee.com"and alias="libCore";

接着，引入 hello 脚本：

include local.`libCore.udf.hello`;select hello() as name as output;

结果为：name: hello world，最后就可以在 select 句式中使用 hello 函数了。

4.6.宏函数/!Macro

标准 SQL 中也有函数，Byzer-lang 的宏函数则是 SQL 层级的。

譬如每次都写完整的 load 语句是一件比较麻烦的事情。可以将其进行封装：

set loadExcel = ''' load excel. {0} where header="true" as {1} ''';!loadExcel ./example-data/excel/hello_world.xlsx helloTable;

宏函数使用类似命令行的方式进行参数传递，使用前缀 ! 进行宏函数的调用。

宏函数也支持命名参数：

set loadExcel = ''' load excel. ${path} where header="true" as ${tableName} ''';!loadExcel _ -path ./example-data/excel/hello_world.xlsx -tableName helloTable;

4.6.1.内置宏函数

4.6.2.信息查看类命令

在 Byzer 中可以通过 !show 命令来查看相关信息

分类	语句	功能描述
系统版本	!show version;	查看当前引擎版本
命令	!show commands;	列出 show 命令支持的所有子命令
表	!show tables;	列出所有的表
表	!show tables from [databaseName];	列出指定 databaseName 中的表
任务	!show jobs;	列出当前正在运行的任务
任务	!show "jobs/[jobGroupId]"; !show "jobs/v2/[jobGroupId]";	列出指定 jobGroupId 的任务详细信息
数据源	!show datasources;	列出当前系统中可用的数据源
数据源	!show "datasources/params/[datasourceName]";	列出指定数据源 datasourceName 的参数列表
API	!show "api/list";	列出 REST API 接口的相关信息
配置	!show "conf/list";	列出支持的配置参数，当前的实现可能不全，以用户手册文档为主
日志	!show "log/[offset]";	查看指定偏移位置的日志
硬件资源	!show resource;	查看当前系统的资源情况
ET	!show et;	列出系统中的 ET 组件
ET	!show "et/[ET Name]";	列出指定的 ET Name 的 ET 组件详情
ET	!show "et/params/[ET Name]";	列出指定的 ET Name 的 ET 组件中的参数列表信息，比如 !show "et/params/RateSampler; 是查看 RateSampler 组件包含的所有参数信息
函数	!show functions;	查看当前引擎版本
函数	!show "function/[functionName]";	查看当前引擎版本

4.6.3.文件操作

在 Byzer 中可以通过 !hdfs 和 !fs 来对文件进行操作 !hdfs 和 !fs 是相同的命令，可以互换使用支持大部分常见的 Hadoop 系统中 HDFS 的查看命令

分类	语句	功能描述
帮助	!hdfs -help; !hdfs -usage;	查看该命令的帮助信息
查看文件（夹）	!hdfs -ls [path];	列出指定路径 path 下的文件和文件夹信息
删除文件（夹）	!hdfs -rmr [path];	删除指定路径 path
复制文件（夹）	!hdfs -cp [source] [destination];	复制 source 文件路径至 destination
移动文件（夹）	!hdfs -mv [source] [destination];	移动 source 文件路径至 destination
移动文件（夹）	!hdfs -mv [source] [destination];	移动 source 文件路径至 destination
重命名文件	!hdfs utils rename "/tmp/.csv" "/tmp/sub" "(\\.csv)$" ".txt"; 或 !fs utils rename _ -source "/tmp/.csv" -target "/tmp/sub" -pattern "(\\.csv)$" -replace ".txt";	重命名指定 source 路径下的文件，将符合正则表达式 pattern 匹配的文件，重命名（或replace）指定的字符串，没有匹配到规则的文件，则保持原样移动到新的目录下
合并文件	!hdfs utils utils getmerge "/tmp/.csv" "/tmp/a.csv" 1; 或 !fs utils utils getmerge _ -source "/tmp/.csv" -target "/tmp/a.csv" -skipNLines 1;	合并指定 source 路径下匹配到的文件，匹配通过通配符 *，将合并后的文件合并单一文件，并移动至 target 指定的目录，skipNLines 参数指定了除了第一个文件暴露前 N 行，其余的文件则去掉前 N 行，一般应用于当合并的文件有 header 时，最后合并的文件不会有重复 header 出现

4.6.4.任务操作

分类	语句	功能描述
查看任务	!show jobs;	列出当前正在运行的任务
查看任务	!show "jobs/[jobGroupId]"; !show "jobs/v2/[jobGroupId]";	列出指定 jobGroupId 的任务详细信息
结束任务	!kill [jobName];	结束指定 jobName 的任务
结束任务	!kill [jobGroupId];	结束指定 jobGroupId 的任务

4.6.5.表操作

分类	语句	功能描述
查看表	!show tables;	列出所有的表
查看表	!show tables from [databaseName];	列出指定 databaseName 中的表
查看 Schema	!desc [tableName];	查看指定 tableName 的 Schema
缓存表	!cache [tableName] [lifecycle];	缓存指定 tableName 的表至内存中，lifecycle 有两个值，分别是 script 和 session； - 其中 script 表示该表的缓存只在当前的脚本有效（即通过 API 一次执行后就失效），用户无需手动进行释放缓存； - 其中 session 表示该表的缓存在引擎的session 中和用户绑定，需要用户手动执行释放后才会被释放一般情况下，为了加快计算，我们建议使用 script 生命周期来进行缓存
释放表缓存	!unCache [tableName];	释放指定 tableName 的表的缓存
定义命令的结果返回为表	!lastCommand named [tableName];	将脚本中上一条执行的宏命令语句的结果定义为临时表 tableName，在后续调用时，可以使用 select 语句来操作上一条宏命令的结果
获取表名称	!lastTableName;	使用该命令跟在一条定义了虚拟临时表的语句后面，在后续的引用中可以通过变量 "${__last_table_name__}" 来获取该虚拟临时表的名称。比如有一条语句 select 1 as a as table1;, 在该语句后执行 !lastTableName;, 在后续的语句可以这样获取表的名称进行操作 select "${__last_table_name__} as tableName as output ，输出为 table
表分区	!tableRepartition _ -i [tableName] -num [partitionNum] -o [outputTableName];	对指定的表 tableName 进行分区，分区数量由 partitionNum 指定，分区后会产生一张新的表 outputTableName
忽略表结果的返回	!emptyTable;	有些时候可能结果集过大或不需要将表结果返回给调用方，那我们就可以在脚本的末尾使用 !emptyTable;

4.6.6.Delta 操作

分类	语句	功能描述
显示帮助	!delta help;	显示帮助信息
查看 Delta 表	!delta show tables;	列出 delta 中的所有表
查看版本历史	!delta history [dbName].[tableName];	列出 delta 中指定 dbName.tableName 的表的历史信息，可以使用反引号将 [dbName].[tableName] 进行 quote 操作
查看 Delta 表信息	!delta info [dbName].[tableName];	查看 delta 中指定 dbName.tableName 的表的信息
文件合并	!delta compact [dbName].[tableName] [version] [fileNum] background;	合并 delta 中指定 dbName.tableName 的表的文件，version 是表的版本信息， fileNum 是合并后的文件数量， background 表示后台执行。以语句 !delta compact demo.table1 100 3 background; 为例，表示对 demo.table1 这张表，在 100 之前的版本的数据文件进行合并，合并后每个目录下只保留 3 个文件

4.6.7.其他命令

分类	语句	功能描述
打印文本	!println '''[contentString]''';	打印文本 contentString
执行文本	!runScript '''[contentString]''' named [tableName];	将文本内容 contentString 作为 byzer 脚本来进行执行，结果命名为 tableName 比如 !runScript ''' select 1 as a as b; ''' named output;
保存文件	!saveFile _ -i [tableName] -o [pathToSave];	如果 tableName 这张表中只有一列，且只有一条记录（即单一 cell），且该 cell 内的内容是 binary 格式，那么我们可以将该 binary 内容保存为一个文件
分析器	!profiler sql '''select 1 as a''';	通过分析器直接执行原生 Spark SQL
分析器	!profiler conf;	通过分析器获取 spark 内核的配置
分析器	`!profiler explain [ SQL	tableName ]`

4.7.Native UDF

Byzer-lang 支持用户使用 Java/Scala 编写 UDF 函数。 Byzer-lang 的一大优势是，随写随用。

register ScriptUDF.`` as arrayLast where lang="scala" and code='''def apply(a:Seq[String])={ a.last }''' and udfType="udf"; select arrayLast(array("a","b")) as lastChar as output;

结果为：lastChar: b

在上面的代码中定义了一个 arrayLast 的函数，该函数的逻辑是获取数组中最后一个值。通过 register 句式注册完成后，就可以在 select 句式中直接使用，效果和内置的函数一样。

4.8.变量设置/Set

Byzer-lang 也支持变量。变量使用 set 进行声明。

比如：

set a="b";

变量可以被应用于 select 句式中：

select "${a}" as a as output;

结果为：a

在 Byzer-lang 中，变量引用主要以 ${} 的方式进行，分支条件表达式则是特例，它以 :VariableName 形式进行引用。

4.9.分支语句/!If|!Else

Byzer-lang 支持高级别的分支语句。

示例如下：

set a = "wow,jack";!if ''' split(:a,",")[0] == "wow" ''';select 1 as a as b;!else;select 2 as a as b;!fi;select * from b as output;

亦或是：

select 1 as a as mockTable;set b_count=`select count(*) from mockTable ` where type="sql" and mode="runtime";!if ''':b_count == 1 '''; select 1 as a as final_table;!else; select 2 as a as final_table;!fi; select * from final_table as output;

结果为：a: 1

4.10.机器学习 —— 模型训练/Train｜Run

load/select 句式可以完成对数据的加载，关联，预处理等相关工作。处理完成后，可以对结果表使用 train 句式进行训练。

train mock_data as RandomForest.`/tmp/models/randomforest` where keepVersion="true" and evaluateTable="mock_data_validate"and `fitParam.0.labelCol`="label"and `fitParam.0.featuresCol`="features"and `fitParam.0.maxDepth`="2";

这句话表示，使用 mock_data 表为训练数据，使用 RandomForest 进行训练，训练的参数在 where/options 子句中申明。得到的模型保存在 /tmp/models/randomforest 中。

4.11.机器学习 —— 模型注册/Register

训练成功得到模型文件后，就可以将模型注册成 UDF 函数供后续将模型应用于批，流，API中。此时可以使用 register 句式完成模型到 UDF 的注册。

register RandomForest.`/tmp/models/randomforest` as model_predict;select vec_array(model_predict(features)) as predicted_value from mock_data as output;

4.12.机器学习 —— 模型预测/Predict

示例代码如下：

predict testData as RandomForest.`/tmp/models/randomforest`;

4.13.机器学习——模型自解释性

模型训练后，可以使用 modelExplain 语法查看模型参数。下面举例说明。首先，训练 2 个随机森林模型，并保存至 /tmp/model 目录。

set jsonStr=''' {"features":[5.1,3.5,1.4,0.2],"label":0.0}, {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.4,2.9,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[4.7,3.2,1.3,0.2],"label":1.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} {"features":[5.1,3.5,1.4,0.2],"label":0.0} '''; load jsonStr.`jsonStr` as mock_data; select vec_dense(features) as features, label as label from mock_data as mock_data_1; -- use RandomForest train mock_data_1 as RandomForest.`/tmp/model` where keepVersion="true" and evaluateTable="mock_data_validate" and `fitParam.0.labelCol`="label" and `fitParam.0.featuresCol`="features" and `fitParam.0.maxDepth`="2" and `fitParam.1.featuresCol`="features" and `fitParam.1.labelCol`="label" and `fitParam.1.maxDepth`="10" ;

完成后，结果显示模型目录分别是 /tmp/model/_model_8/model/1 和 /tmp/model/_model_8/model/0

然后，查看模型参数。

load modelExplain.`/tmp/model/` where alg="RandomForest" and index="8" as output;

这里，结合 /tmp/model 和 index="8" ，系统读取 /tmp/model/_model_8 的模型，并以随机森林算法解释之。上面的语句等价于

load modelExplain.`/tmp/model/_model_8` where alg="RandomForest" as output;

查看features的对模型影响的权重：

--查看featureImportance --(4,[0,2],[0.24854914169719383,0.7514508583028062]) load modelExplain.`/tmp/model/_model_8` where alg="RandomForest" as output; select value as feature_Importance from output where name ='featureImportance' as fi;

4.14.Byzer-python 支持

Byzer 通过 Byzer-python 支持 Python 脚本。如果用户在 Byzer Notebook 中使用，将会更加易用。

下面展示的是一段纯 Byzer-lang 的代码：

select 1 as a as mockTable;-- 设置 python 输出模式!python conf "schema=st(field(a,long))";-- 设置执行 python 代码的虚拟环境!python env "PYTHON_ENV=source /opt/miniconda3/bin/activate ray1.8.0"; run command as Ray.`` where inputTable="mockTable"and outputTable="newMockTable"and code=''' from pyjava.api.mlsql import RayContext ray_context = RayContext.connect(globals(),None) newrows = [] for row in ray_context.collect(): row["a"] = 2 newrows.append(row) context.build_result(newrows) ''';select * from newMockTable as output;

在 Notebook 中，语法会得到简化，可以在单独的 cell 里写 Python 脚本。

4.14.1.Shema数据类型

Schema 表达

当我们使用 Python 代码创建新表输出的时候，需要手动指定 Schema. 有两种情况：

如果返回的是数据，可以通过设置 schema=st(field({name},{type})...) 定义各字段的字段名和字段类型；
如果返回到是文件，可设置 schema=file；

设置格式如下：

!python conf "schema=st(field(_id,string),field(x,double),field(y,double))";

schema 字段类型对应关系：

Python 类型	schema 字段类型	例（Python 数据：schema 定义）
int	long	{"int_value": 1} ："schema=st(field(int_value,long))"
float	double	{"float_value": 2.1} ："schema=st(field(float_value,double))"
str	string	{"str_value": "Everything is a table!"} ："schema=st(field(str_value,string))"
bool	boolean	{"bool_value": True} ："schema=st(field(bool_value,boolean)"
list	array	{"list_value": [1.0, 3.0, 5.0]}："schema=st(field(list_value,array(double)))"
dict	map	{"dict_value": {"list1": [1, 3, 5], "list2": [2, 4, 6]}} ："schema=st(field(dict_value,map(string,array(long))))"
bytes	binary	{"bytes_value": b"Everything is a table!"} ："schema=st(field(bytes_value,binary))"