DolphinDB计算节点使用指南

　　导读
　　为了提升DolphinDB在高并发读写场景下的性能与稳定性，DolphinDB在架构上引入了计算节点（computenode）。计算节点接管了数据节点的部分职能，负责响应客户端的请求并返回结果。在架构层面，将集群的计算与存储进行分离，保证数据节点的软硬件资源有效服务于IO过程，从而提升集群写入性能与稳定性。同时DolphinDB集群管理者可以根据业务需求和集群负载情况，对计算资源和存储资源独立扩展。
　　本教程将重点介绍计算节点的特性、架构设计、应用实践及运维管理等方面内容，通过原理解析、案例分析来讲解如何部署和使用计算节点。面向读者架构师（15章）DolphinDBDBA（第1、2、4章）开发工程师（第1、3章）运维工程师（第1、4章）第1章概览1。1为什么引入计算节点
　　数据节点可以同时承担计算任务和数据读写任务的执行，对于一些计算任务不重的场景来说，数据节点兼任计算任务没有问题，但是对于计算逻辑较复杂，并发度较高的计算密集场景下，可能会有以下影响：对于数据写入的影响：由于密集计算任务会大量消耗CPU和内存资源，而数据写入时需要CPU来进行数据的预处理和压缩，因此会影响数据写入性能。对于数据读取的影响：若大量内存被计算任务占用，那么读取数据时，可能因内存不足导致OOM异常。
　　DolphinDB自版本1。30。14，2。00。1开始支持计算节点，能够有效地解决上述问题。1。2计算节点简介1。2。1核心特性
　　计算节点，顾名思义，即执行计算任务的节点，相比于数据节点，有以下2个核心特性：不存储分布式表数据
　　不存储分布式表数据、元数据文件，是一个轻状态的服务组件。计算节点的启停流程十分精简，任何故障都可以通过重启节点来进行快速修复。同时，计算节点可以进行快速地扩缩容。有关计算节点的启停步骤，参考：启动数据节点和计算节点
　　如下图所示3个数据节点、3个计算节点的集群：
　　图11
　　图中C1CNODE1至C1CNODE3为计算节点，P1DNODE1至P3DNODE1为数据节点。
　　从文件系统来简单验证下计算节点的无状态特性。查看数据节点volume配置：查看存储卷selectfrompnodeRun（getConfig｛volumes｝）P1DNODE1ssdssd0wfHuangvolumesP1DNODE1P2DNODE1ssdssd0wfHuangvolumesP2DNODE1P3DNODE1ssdssd0wfHuangvolumesP3DNODE1
　　到操作系统去查看数据节点P1DNODE1的volume目录：
　　图12
　　可看到P1DNODE1作为数据节点，管理着事务undoredo日志文件，和大量的数据块CHUNKS文件。但计算节点不需要配置volume，查看C1CNODE1的默认data目录：
　　图13
　　可以看到计算节点C1CNODE1不存储任何的分布式数据文件。负责计算
　　在分布式表读写场景中，由数据节点负责map阶段，计算节点负责mergereduce阶段（详见1。2。2章节）。在纯内存对象（常规内存表、分区内存表、流数据表等）计算场景中，由计算节点独立执行。
　　执行一个分布式表查询SQL，从文件层面来观察计算节点在数据查询发挥的作用。
　　查看计算节点C1CNODE1打开的文件（openfiles）：〔appadmincomputevolumes〕lsofppidOfComputenodeskiplibraryfilescomputew38563appadmin19uIPv43191810090t0TCPcompute：8961192。192。168。2：56780（ESTABLISHED）computew38563appadmin20uIPv43191035500t0TCPcompute：8961192。192。168。2：56786（ESTABLISHED）computew38563appadmin22uIPv43190955500t0TCPcompute：8961192。192。168。2：34842（ESTABLISHED）。。。skip20lines。。。。。。computew38563appadmin44uIPv43193375130t0TCPcompute：42566192。192。168。2：8961（ESTABLISHED）
　　查看数据节点P1DNODE1打开的文件：〔appadminsupport4log〕lsofppidOfDatanodecomputew99763appadmin33uIPv4494517660t0TCPsupport4：8961192。192。168。1：42474（ESTABLISHED）computew99763appadmin36uIPv4494556180t0TCPsupport4：8961192。192。168。1：42562（ESTABLISHED）computew99763appadmin39rREG8，17272358102177313580ssdssd0wfHuangvolumesP1DNODE1CHUNKStestcompute20220102hQEmachines2100000058computew99763appadmin40rREG8，17272358117177313646ssdssd0wfHuangvolumesP1DNODE1CHUNKStestcompute20220106hQEmachines2100000062computew99763appadmin41rREG8，17272358131177313802ssdssd0wfHuangvolumesP1DNODE1CHUNKStestcompute20220227hQEmachines2100000072computew99763appadmin42rREG8，17272358111177313644ssdssd0wfHuangvolumesP1DNODE1CHUNKStestcompute20220109hQEmachines2100000060。。。。。。skiphundredsoflines。。。。。。
　　可以观察到计算节点C1CNODE1打开了大量的TCP连接符，而数据节点P1DNODE1除了TCP连接符外，还打开了大量的磁盘文件。
　　图14
　　如上图所示，计算节点computenode1负责解析client的请求，并协调数据节点datanode〔13〕读写数据（即map，分布式执行过程）。接收到datanode〔13〕的数据后，经过reducemerge过程，将最终结果返回给客户端。存算分离的架构，使得数据节点的物理资源得到隔离，写入和读取数据更为稳定、高效。1。2。2通过一个SQL查询理解计算节点
　　当客户端向计算节点发起一个sql查询：pt1loadTable（database（dfs：szstock），sztrade）select〔HINTEXPLAIN〕securityID，min（price）asLowPx，max（price）asHighPx，sum（tradeQty）asvol，first（price）asOpenPx，last（price）asLastPx，sum（tradeQtyprice）asvalfromptwhereexecTypeFgroupbytradedate，securityID，minute（temporalParse（tradetimeSTRING，yyyyMMddHHmmssSSS））asminuteorderbytradedate，securityID，minute
　　通过〔HINTEXPLAIN〕获取执行计划，整个sql的执行过程依次为parse，map，merge，reduce四个部分。｛measurement：microsecond，explain：｛from：｛cost：2｝，map：｛partitions：｛local：0，remote：320｝，cost：5189927，detail：｛。。。skipmanylines。。。｝，merge：｛cost：446588，rows：8522241，detail：｛。。。skipmanylines。。。｝｝｝，reduce：｛sql：select〔98307〕from702a9832b17f0000orderbytradedateasc，securityIDasc，minuteasc，explain：｛sort：｛cost：296783｝，rows：8522241，cost：746374｝｝，rows：8522241，cost：6691112｝｝
　　其中，queryparse、merge、reduce过程发生在计算节点，map阶段则并行发生于各数据节点。
　　图15
　　整个执行过程消耗资源具体如下表。
　　执行步骤
　　发生节点
　　消耗资源
　　时间（us）
　　queryparse
　　computenode
　　cpu
　　2hrmap
　　datanode3
　　IO，network，cpu，memory
　　5189927hrmerge
　　computenode
　　cpu，network，memory
　　446588hrreduce
　　computenode
　　cpu，network，memory
　　746374hr该SQL总共涉及300个分区，各分区记录数在60280万不等。mergereduce阶段分布处理了850万的数据。由C1CNODE1承担的mergereduce阶段总耗时占比为
　　（446588746374）669111217。8
　　上述例子map阶段where，groupby算子大大减少了数据量，因此计算节点的算量不大，如果是读取数据类的场景，比如数据导入导出，此比例会大大增加。
　　执行过程中，计算节点与数据节点的内存消耗情况如下图所示（node33：计算节点，node3436：数据节点，下同）。
　　图16
　　可以看出在map阶段在数据节点，是分布式、并行执行的。在mergereduce过程中计算节点的内存消耗逐渐增加。1。3引入计算节点的好处1。3。1保证写入数据的稳定性
　　在数据节点上执行复杂业务计算（如因子计算、机器学习等），会占用大量的内存。使得cacheEngine无法获取足够多的内存，而发生cacheEngine内存溢出等问题，导致写入延迟，甚至失败。
　　ChunkCacheEnginecantreclaimanymemoryaftertrying13times。
　　Transationfailed，cacheEngineoutofmemory。
　　另外，使用TSDB引擎，数据刷盘时需进行排序，需要消耗大量的CPU资源以及排序buffer。若CPU、内存资源不足时，TSDB的写入性能会下降。
　　通过在架构上引入计算节点，采用计算与存储分离的架构设计，可以有效地保证集群数据写入的稳定性。1。3。2降低故障平均修复时间
　　使用计算节点的集群能显著地降低故障平均修复时间，尽可能地减少对业务的影响。核心原因是由于计算节点不管理分布式数据，重启可以在数秒内完成。当某个计算节点出现拥塞、无法响应等情况，且无法热修复的，重启即可使其恢复正常服务。启动流程对比
　　图17
　　如图，当数据节点启动时，流程中的蓝色标记部分：回放redo日志，恢复损坏的chunk等包含了大量的IO操作，实际生产环境中耗时为分钟级（具体视数据量大小而定）。而计算节点流程十分精简，可以在秒级完成启动。1。4计算节点安装
　　现用如下4台服务器部署集群：
　　P1：192。193。168。2
　　P2：192。193。168。3
　　P3：192。193。168。4
　　C1：192。193。168。1
　　其中服务器P1P3用于部署数据节点，C1用于部署计算节点。1。4。1编写配置文件
　　在节点列表文件cluster。nodes中编写计算节点及代理节点。推荐计算节点以CN（取computenode首字母）标识，数据节点以DN（取datanode首字母）标识，如：localSite，mode192。193。168。2：8990：controller1，controller192。193。168。3：8990：controller2，controller192。193。168。4：8990：controller3，controller192。193。168。2：8960：P1agent，agent192。193。168。3：8960：P2agent，agent192。193。168。4：8960：P3agent，agent192。193。168。1：8960：C1agent，agent192。193。168。2：8961：DN1，datanode192。193。168。3：8961：DN2，datanode192。193。168。4：8961：DN3，datanode192。193。168。1：8961：CN1，computenode192。193。168。1：8966：CN2，computenode192。193。168。1：8967：CN3，computenode
　　在配置文件cluster。cfg中，需要使用前缀来区分数据节点与计算节点的配置项。CN。workerNum64CN。localExecutors63CN。maxConnections3000CN。maxMemSize256CN。dataSync0DN。maxConnections512DN。maxMemSize128DN。workerNum32DN。localExecutors31DN。chunkCacheEngineMemSize6DN。dataSync1DN。redoLogDirssdssd0wfHuangvolumesredoLogDN。volumesssdssd0wfHuangvolumes，ssdssd1wfHuangvolumes，ssdssd2wfHuangvolumes
　　CN匹配CN开头的计算节点，DN匹配DN开头的数据节点。以此保证计算节点与数据节点的配置项不会互相干扰。
　　注：计算节点必须配置dataSync0。1。4。2启动计算节点
　　参考集群安装模式部署并启动agent、controller节点。启动完成后后，使用web启动计算节点。
　　图18第2章架构设计2。1计算节点的设计和硬件规划
　　依据上文解析，计算节点主要负责接收、响应请求以及内存计算，数据节点主要负责数据存储管理。在集群设计与规划时，生产环境最小硬件配置推荐如下：
　　组件
　　CPU
　　内存
　　网络
　　硬盘
　　datanode
　　8核
　　64GB
　　10Gb网卡
　　ssdhdd，500GB建议元数据存ssd，分区数据存hdd
　　computenode
　　16核
　　128GB
　　10Gb网卡
　　hdd，200GB
　　计算节点更注重cpu、内存、网卡性能。而数据节点更注重iops，存储通道与容量。集群拓扑结构可根据应用场景与需求参考下文设计。2。2单计算节点集群架构
　　图21
　　单一的计算节点，可以做为请求统一入口。此时拓扑简单，部署方便。2。3高可用计算节点集群架构
　　图22
　　DolphinDB数据库层面，部署多个计算节点，扩展集群计算容量，并实现高可用。
　　从计算节点发起DFS表读写请求时，可以容忍部分数据节点实例级别的故障（比如某个数据节点的服务器断电），其崩溃恢复策略如下：写
　　从计算节点写入数据时，如果某个数据节点发生了崩溃，事务管理器会回滚当前失败的事务。并以一个新的事务，重试数据写入，整个过程对于客户端程序是透明无感知的。
　　图23读
　　DolphinDB默认采用2副本（确保强一致性），当计算节点发现部分子任务失败时，会根据可用chunk副本的分布，重新调度新的subTask到对应数据节点中去，实现任务的崩溃恢复。
　　图24
　　在中间件层面，可以引入负载均衡组件（如haproxy，F5等）来实现集群统一访问接口。2。4流计算节点
　　生产环境中，流数据对时延要求非常高。在多人、多业务场景使用集群时，数据节点由于本身还是承担着数据读写操作，不可避免地在一定程度上影响流计算。在设计上，可以将流计算相关业务单独地部署至某计算节点实例。
　　图25
　　图示，使用computenode3作为流计算节点，专门负责流计算业务。在另外两个计算节点上进行批量读写，通过计算节点之间的Sharenothing架构特性，有效保证流计算的低时延。2。5基于计算节点实例的资源隔离
　　在生产环境中，若有多人同时使用DolphinDB集群，需要一定的资源隔离机制来保障核心业务稳定运行。可以在架构层面，配合负载均衡策略，实现基于计算节点的资源隔离。
　　图26
　　以haproxy为例，在haproxy。cfg中设置2个监听组ddbcluster
　　核心业务使用计算资源组，总共6000连接，2个计算节点，高可用。adhocsingle
　　用于即席查询的计算节点，配置为单节点，限制1000连接，用于分析师，研究员等对数据做一些探索和即席查询。listenddbclusterbind0。0。0。0：5000modetcpbalanceleastconnservercompute1192。168。168。1：8961maxconn3000checkservercompute2192。168。168。1：8966maxconn3000checklistenadhocsinglebind0。0。0。0：10090modetcpbalanceleastconnservercompute3192。168。168。1：8967maxconn1000check第3章应用场景
　　所有数据节点的计算任务、代码都可以无缝迁移到计算节点。集群管理与使用者应当避免直接访问数据节点，转而使用计算节点作为查询与写入的入口。具体演示数据读取、写入、流计算、机器学习。3。1数据读取
　　使用pythonapi从计算节点读取某一只股票的历史交易数据。importDolphinDBasddbdefgetData（date，securityID，rowNum1000000）：sddb。session（）s。connect（computenode1，8961，admin，123456）trades。loadTable（tableNamesztrade，dbPathdfs：szstock）rsttrade。select（）。where（tradedateddate）。where（securityIDssecurityID）。where（execTypeF）。limit（rowNum）。toDF（）s。close（）returnrstifnamemain：print（getData（20200102，000014））3。2数据写入
　　将历史行情数据导入DolphinDB，具体操作请参考国内股票行情数据导入实例。导入2020年1月数据，共计82G数据时，集群计算节点，数据节点消耗情况如下：cpumemory
　　图31io
　　图32
　　在执行导入过程中，计算节点cpu均值在15，峰值达到20，内存消耗约2。8G。3。3流计算
　　流计算的核心诉求是时延，如果采用数据节点作为流计算节点，必不可免地会受到分布式表读取与写入的影响。配置计算节点成为流计算节点，可以有效地保障CPU，memory，io，worker线程等软硬件资源服务于流计算，从而保证低时延。并且由于计算节点之间是互不影响的，推荐在架构上设计部分计算节点作为流计算节点，其他计算节点用于分析与查询。3。3。1配置计算节点
　　设计CN1作为流计算的订阅和发布节点，配置文件如下：publishCN1。persistenceDirssdssd0persistDirC1CNODE1CN1。persistenceWorkerNum4CN1。maxPubConnections64subscribeCN1。maxSubConnections64CN1。subPort8970CN1。subExecutors163。3。2创建与订阅流表
　　在计算节点CN1上创建流表：tstreamTable（1：0，symprice，〔STRING，DOUBLE〕）enableTableShareAndPersistence（t，tickStream）
　　订阅并编写实时因子计算：defsumdiff（x，y）｛return（xy）（xy）｝factor1ema（1000sumdiff（ema（price，20），ema（price，40）），10）ema（1000sumdiff（ema（price，20），ema（price，40）），20）resulttable（1000：0，symfactor1，〔STRING，DOUBLE〕）rsecreateReactiveStateEngine（namereactiveDemo，metricsfactor1，dummyTabletickStream，outputTableresult，keyColumnsym）subscribeTable（tableNametickStream，actionNamefactors，offset1，handlertableInsert｛rse｝）
　　往CN1模拟写入消息（同时另一个计算节点CN2执行导入数据作业）n2000000datatable（take（000001。SH，n）assym，rand（10。0，n）asprice）tickStream。append！（data）getStreamingStat（）。subWorkers
　　查看集群的资源消耗情况：
　　图33
　　可以看到CN2，DN13在消耗相对较高的内存和CPU时，CN1并没有受到数据导入的影响，使得流计算的时延得到有效保证。3。4机器学习
　　机器学习是计算密集型的场景，模型的训练过程会消耗大量的CPU、内存资源。将机器学习的作业部署至某个计算节点，可以避免对数据写入、读取类任务的负面影响。例如在CN2上，使用adaBoost训练股票年波动率：deftranAdaBoost（TrainData）｛dbdatabase（，HASH，〔SYMBOL，10〕）p10TranDatadb。createPartitionedTable（tableTrainData，partitionColumnsSecurityID）p10TranData。append！（TrainData）modeladaBoostRegressor（sqlDS（selectfromp10TranData），yColNametargetRV，xColNamesBASDI0DI1DI2DI3DI4PressRV，numTrees30，maxDepth16，losssquare）saveModel（model，hdddatafinancemodeladaBoost。txt）｝jobIdadaBoostjobDescadaBoosttrainsnapsubmitJob（jobId，jobDesc，tranAdaBoost，Train）
　　完整代码及过程请参考实时实际波动率预测。使用10个线程并行训练，加速训练过程。
　　图34
　　在CN2训练过程中，其他节点资源未受影响。同时如果想加速机器学习训练过程，可以使用更多的线程，为CN2额外配置一些CPU、内存资源，在资源管理上非常灵活。第4章运维4。1集群监控
　　引入计算节点的集群，在运维监控层面，需要以存储与计算分离的架构来进行监控。对于数据节点、计算节点的监控侧重点不同。具体来说，计算节点需要关注客户端请求执行情况的相关性能：包括平均响应时间、连接数、内存、CPU使用率、网卡带宽使用率等；而数据节点更需要关注磁盘空间、IOPS、读写MBs等IO相关指标。4。1。1监控软件部署
　　通过部署第三方软件：Grafana，Prometheus实现集群的管理与监控，具体流程参考Grafana系列部署。4。1。2监控面板
　　提供计算节点与数据节点监控dashboard，是基于存储与计算分离架构设计的一套监控面板。已上传至Grafana官方仓库，可以在快速完成集群的运维监控部署工作。
　　点击Import输入对应dashboardID，点击Load即可完成导入。
　　图414。2查询监控与SESSION管理4。2。1session
　　计算节点监控的关键运行时参数主要是内存、连接数。如何避免个别session占用过多内存，而导致其他sessionOOM，是运维管理上的重点。内存管理
　　在内存紧张时，通过getSessionMemory视图进行会话管理，回收空闲会话：selectuserId，sessionId，memSize10241024asMB，localtime（createTime）ascreateTime，localtime（lastActiveTime）aslastActiveTime，（now（）localtime（lastActiveTime））1000asidleSecondsfromgetSessionMemoryStat（）orderbymemSizedesc
　　通过closeSessions（sessionId）来关闭对应会话并回收内存。
　　图42
　　图示，test用户已经空闲15分钟了，视情况可以关闭此session，并回收1000MB的内存。
　　2。连接管理
　　在Linux中，DolphinDB的一个connection对应一个TCP连接。在物联网场景下，需要调整连接数（每个tcp连接消耗4K内存，对操作系统消耗不大），以应对高QPS需求，建议将计算节点的配置设置为maxConnections5000及以上。
　　Thenumberofincomingconnectionshasreachedthelimitof512
　　当日志出现以上信息时，说明连接已经不足，检查程序是否存在连接泄露的情况。应急处理请使用closeSessions（）清理一些空闲会话以回收连接，并增加maxConnections。4。2。2慢查询
　　慢查询通常会引起系统雪崩，需要重点关注。DolphinDB提供一个joblog日志文件，记录DFS表相关查询日志，默认在log目录下，名称为nodeAliasjob。log。监控top10sqljobLogFilegetConfig（jobLogFile）queryLogloadText（jobLogFile）selectnode，userId，sessionId，jobDescassqlText，startTime，endTime，（endTimeDATETIMEstartTimeDATETIME）ascostSecondsfromqueryLogorderby（endTimeDATETIMEstartTimeDATETIME）desclimit10
　　图43
　　如图，根据sqlText，userId，运行时间及时长，综合分析处置慢查询，比如限制一些用户进行较大规模的查询，以及对部分查询效率和速度较低的SQL语句进行优化。4。3参数设置
　　计算节点不需要设置DFS表存储相关的参数，包括：volumes，chunkCacheEngineMemSize，tsdbCacheEngineSize，redoLogDir等。计算节点的核心参数包括：MaxMemSize，MaxConnections，workerNum，localExecutor，maxQueryResultLimit等。具体设置请参考手册参数配置。4。4计算节点扩缩容
　　相比于数据节点，计算节点扩缩容非常简单快捷，支持在线扩缩容。4。4。1扩容在新机器上部署和配置agent
　　拷贝原机器上的agent部署包到新机器，并修改agent。cfg，如modeagentworkerNum4localExecutors3localSite192。192。168。4：8950：C4agentcontrollerSite192。192。168。4：8990：controller3sites192。192。168。4：8950：C4agent：agent，192。192。168。2：8990：controller1：controller，192。192。168。3：8990：controller2：controller，192。192。168。4：8990：controller3：controllerlanCluster0
　　2。在web集群管理界面修改NodesSetup配置，新增computenode和agent。
　　图44
　　回到web管理界面，可以看到已经新增了一个agent4并且是未启动状态。
　　3。启动新节点的agentnohup。DolphinDBconsole0modeagenthomedataconfigconfigagent。cfglogFilelogagent。log
　　注：
　　如果在已有机器上扩展一个计算节点实例，只需在web集群管理页面的任意一个数据计算节点执行命令：addNode（192。168。100。33，8965，C1NODE5，，computenode）
　　新计算节点命名应遵守部署文档中提到的命名规则，以确保cluster。cfg能赋予新的计算节点正确的配置。4。4。2缩容停止计算节点
　　在web集群管理页面停止对应计算节点，等待其安全关闭，通常需要数秒。
　　2。修改集群配置
　　在nodesSetup中删除对应的computenode节点。第5章总结
　　计算节点不存储分布式数据，使用计算节点可以有效地提升集群的写入稳定性，缩短故障修复时间。计算节点之间相互隔离，架构设计者可根据应用场景，灵活地部署业务至各计算节点。推荐将流计算、机器学习部署至计算节点。在运维管理上，计算节点可以快速重启、方便地在线扩缩容。