建设网站的风险6,网站建设 采集,坪地网站建设价格,创意设计生活用品1. 前言
DolphinDB 的线程模型较为复杂#xff0c;写入与查询分布式表都可能需要多个类型的线程。通过了解 SQL 查询的全过程#xff0c;可以帮助我们了解 DolphinDB 的线程模型#xff0c;掌握 DolpinDB 的配置#xff0c;以及优化系统性能的方法。
本教程以一个分布式 …1. 前言
DolphinDB 的线程模型较为复杂写入与查询分布式表都可能需要多个类型的线程。通过了解 SQL 查询的全过程可以帮助我们了解 DolphinDB 的线程模型掌握 DolpinDB 的配置以及优化系统性能的方法。
本教程以一个分布式 SQL 查询为例介绍 DolphinDB 分布式数据库查询过程的数据流以及其经历的各类线程池。
2. DolphinDB 的主要节点类型
agent代理节点
负责执行控制节点发出的启动和关闭数据节点或计算节点的命令。在一个集群中每台物理服务器有且仅有一个代理节点。
controller控制节点
负责收集代理节点、数据节点和计算节点的心跳监控每个节点的工作状态管理分布式文件系统的元数据和事务。
data node数据节点
既可以存储数据也可以用于数据的查询和计算。
compute node计算节点
只用于数据的查询和计算。计算节点应用于计算密集型的操作包括流计算、分布式关联和机器学习等场景。计算节点不存储数据但可以通过 loadTable 加载数据进行计算。通过在集群中配置计算节点可以将写入任务提交到数据节点、所有计算任务提交到计算节点实现存储和计算的分离。
1.30.14/2.00.1 版本开始支持计算节点。 3. DolphinDB 线程类型
3.1 worker 线程
常规交互作业的工作线程。每个节点都存在 worker 线程可以分为以下几类 ZeroWorkerFirstWorkerSecondWorkerThirdWorkerForthWorkerFifthWorkerSixthWorker客户端提交至节点的作业为 0 级由 ZeroWorker 处理。根据作业所涉及到的分区ZeroWorker 将其分解为多个子任务。其中本地节点上的子任务由 ZeroWorker 与 FirstWorker 并行执行。需要由远程节点执行的子任务则降低为 1 级并通过 remoteExecutor 发送到对应节点上的 FirstWorker 处理。以此类推若某个级别的子任务需要进一步拆解则拆分出来的由远程节点执行的子任务降低一级发送至远程节点上对应层级的 worker 处理。ZeroWorker, FirstWorker, SecondWorker 的线程数量由配置参数 workerNum 决定分别为 workerNum, workerNum-1, workerNum*0.75。其余层级的 work 线程数量为上级的 1/2 向上取整最小个数为 1。配置参数 workerNum 默认值为机器上的 CPU 核数最大值不超过 license 中的最大核数。
3.2 remote executor 线程
将远程任务发送到远程节点的线程。在非 single 模式的节点上可以通过配置参数 remoteExecutors 配置线程个数。默认值为集群中节点个数和本地 worker 的较小值。
3.3 asynchronous remote executor 线程
接收对其他节点发起的远程调用Remote Procedure Call, RPC任务的线程。将收到的远程调用任务放到 remote executor 的任务队列中。每个非 single 模式的节点上有且仅有 1 个该线程。
3.4 remote task dispatcher 线程
在远程调用出错需要重试时或者一个被关闭的连接上仍有未完成的任务时这些任务会先放到一个队列里由 remote task dispatcher 线程从这个队列取任务并重新交由 asynchronous remote executor 线程去发起远程调用。
3.5 batch job worker 线程
执行批处理作业任务的工作线程批处理作业是指通过函数 submitJob 、submitJobEx 、submitJobEx2提交的作业。其上限通过配置参数 maxBatchJobWorker 设置。如果 maxBatchJobWorker 未设置其默认值是配置参数 workerNum 的值。该线程在任务执行完后若闲置 60 秒会被系统自动回收不再占用系统资源。
3.6 web worker 线程
处理 HTTP 请求的工作线程。由配置参数 webWorkerNum 配置默认为 1最大值为 CPU 内核数。
3.7 dynamic worker manager 线程和 dynamic worker 线程
dynamic worker 是动态工作线程是 worker 线程的补充。dynamic worker manager 是创建 dynamic worker 的线程每个节点有且仅有 1 个该线程。当所有的 worker 线程被占满时再有新任务到来时通过 dynamic worker manager 线程创建 dynamic worker 线程来执行新任务。根据系统并发任务的繁忙程度总共可以创建三组动态工作线程每一个级别可以创建 maxDynamicWorker 个动态工作线程。maxDynamicWorker 的默认值为配置参数 workerNum 的值。动态工作线程在任务执行完后若闲置 60 秒则会被系统自动回收不再占用系统资源。
3.8 infra worker 线程
基础设施处理线程。开启高可用后用于接收 raft 心跳汇报的线程防止集群负载大时心跳信息无法及时汇报。默认自动创建 2 个该线程,。
3.9 urgent worker 线程
处理紧急任务。只接收一些特殊的系统级任务譬如登录、取消作业等。由配置参数 urgentWorkerNum 配置默认值为 1最大值为 CPU 内核数。
3.10 block IO worker 线程
执行对硬盘读写任务的线程。由配置参数 diskIOConcurrencyLevel 配置默认值为 1。
4. 一次 SQL 查询的线程经历
SQL 查询可提交到集群中任一计算节点或数据节点。获得请求的节点作为该查询的协调节点。
下面以 API 向协调节点发起一次 SQL 查询为例讲述整个过程中所调度的所有线程。 step1DolphinDB 客户端向协调节点数据节点或计算节点发起数据查询请求
以协调节点为数据节点为例发起一次聚合查询查询语句如下
select avg(price)
from loadTable(dfs://database, table)
where date between 2021.01.01 : 2021.12.31
group by date
假设上述聚合查询语句总共涉及 300 个分区的数据且正好平均分配在三个数据节点即每个数据节点包含了 100 个查询的分区数据。
DolphinDB 客户端将查询请求进行二进制序列化后通过 TCP 协议传输给 datanode1。
step2datanode1 收到查询请求
datanode1 收到客户端的查询请求后将分配 1 个 ZeroWorker 线程对内容进行反序列化和解析。当发现内容是 SQL 查询时会向 controller 发起请求获取跟这个查询相关的所有分区的信息。整个 SQL 查询执行完毕前这个 ZeroWorker 线程会被一直占用。
step3controller 收到 datanode1 的请求
controller 收到 datanode1 的请求后将分配 1 个 FirstWorker 线程对内容进行反序列化和解析准备好本次 SQL 查询涉及的数据分区信息后由该 FirstWorker 线程序列化后通过 TCP 协议传输给 datanode1。controller 的该 FirstWorker 线程完成该工作后将从请求队列中获取下一个请求。
step4datanode1 收到 controller 返回的信息
datanode1 的 ZeroWorker 收到 controller 返回的信息后由本节点下的 1 个 FirstWorker 线程对内容进行反序列化和解析。
得知本次 SQL 查询涉及的数据分区信息后将位于本节点的分区数据计算任务添加到本地任务队列此时本地任务队列会产生 100 个子任务。
同时把需要使用远程节点 datanode2 与 datanode3 的分区数据的计算任务以任务包的方式发送到远程任务队列。远程任务队列会被添加 2 个远程任务分别打上 datanode2 和 datanode3 的标志。
step5 (1)本地 worker 消费本地任务队列
此时datanode1 中的 ZeroWorker 线程和 FirstWorker 线程会同时并行消费本地任务队列的子任务。
step5 (2)、(3)本地 remote executor 发送远程任务至远程节点
同时remote executor 线程将远程任务队列的内容序列化后通过 TCP 协议分别发送到 datanode2 和 datanode3。
step6 (1)、(2)远程节点收到远程任务
datanode2 和 datanode3 收到远程任务后将分配 1 个 FirstWorker 线程对内容进行反序列化和解析并将计算任务发送到本地任务队列此时 datanode2 和 datanode3 的本地任务队列各会产生 100 个子任务。
step7 (1)、(2)远程节点 FirstWorker 消费本地任务队列
此时datanode2 和 datanode3 上的 FirstWorker 线程会并行消费本地任务队列的子任务。
step8 (1)、(2)远程节点返回中间计算结果至 datanode1
当 datanode2 和 datanode3 涉及的计算任务完成后分别得到了本次 SQL 查询的中间计算结果由一直占用的 FirstWorker 线程对内容进行序列化后通过 TCP 协议传输给 datanode1。
step9datanode1 计算最终结果并返回给客户端
datanode1 接收到 datanode2 和 datanode3 返回的中间计算结果后由一直占用的 ZeroWorker 线程对内容进行反序列化然后在该线程上计算出最终结果并在序列化后通过 TCP 协议传输给客户端。
DolphinDB 客户端接收到 datanode1 返回的信息后经过反序列化显示本次 SQL 查询的结果。
协调节点为数据节点和计算节点的区别
数据节点可以存储数据计算节点不能存储数据。但计算节点解析客户端的 SQL 查询后从 controller 拿到本次 SQL 查询涉及的数据分区信息会将所有数据查询任务都分配到数据节点执行得到每个数据节点返回的中间结果最后调度计算节点的 ZeroWorker 线程计算最终结果并返回给客户端。当实时写入的数据量非常大时建议配置计算节点将所有 SQL 查询都提交到计算节点实现存储和计算的分离减轻数据节点的计算工作负担。
5. 优化建议
通过分析上述的线程经历可以发现本次 SQL 查询一共发生了 8 次 TCP 传输其中 2 次是 DolphinDB server 和 DolphinDB client 之间的传输。如果查询结果的数据量比较大同时又对查询结果的延时性比较敏感可以选择如下 6 个优化方向
集群节点之间以及节点和客户端之间的通信推荐使用万兆以太网。优化线程配置参数。(详见本文后续章节)增加每个节点的物理磁盘的数量。更多的磁盘可以更快速地并行读取多个分区的数据。SQL 语句优化where 条件添加分区字段的信息过滤起到分区剪枝的目的避免全表扫描。在查询数据量较大时可对 API 查询结果进行数据压缩提高传输效率。开启压缩后从 server 下载的数据即为压缩后的数据。Java 代码示例如下。
//API 建立 connection 的时候将第三个参数 compress 设置为 true 即可开启压缩
DBConnection connection new DBConnection(false, false, true);
connection.connect(HOST, PORT, admin, 123456);
BasicTable basicTable (BasicTable) connection.run(select * from loadTable(\dfs://database\, \table\));
增加 license 限制的 CPU 核心数和内存大小提升系统的并发处理能力。
6. 不同类型线程与配置参数的关系
线程类型参数配置默认配置配置优化建议workerworkerNumCPU的内核数license 限制的 CPU 核心数与物理机 CPU 核心数两者的最小值remote executorremoteExecutors1集群的节点数-1batch job workermaxBatchJobWorkerworkerNumlicense 限制的 CPU 核心数与物理机 CPU 核心数两者的最小值web workerwebWorkerNum1推荐 webWorkerNum 配置为 4大多情况下很少通过 web 与 DolphinDB 节点交互的方式提交查询任务dynamic workermaxDynamicWorkerworkerNum采用默认值infra workerinfraWorkerNum2采用默认值urgent workerurgentWorkerNum1采用默认值block IO workerdiskIOConcurrencyLevel1对于 hdd 磁盘推荐 diskIOConcurrencyLevel 设为对应节点下通过 volumes 参数配置的磁盘个数对于 ssd 磁盘推荐 diskIOConcurrencyLevel 0
备注
如果是单节点 single 模式或者是单数据节点集群不需要配置 remoteExecutors 的值。
7. 总结
通过本文介绍可以看出DolphinDB的线程模型主要由 worker、remote executor、batch job worker、web worker、dynamic worker、infra worker、urgent worker、block IO worker 组成。在一次 SQL 查询会发生多次 TCP 传输用户可以根据服务器的具体情况进行合理配置线程参数从而降低查询延时。
