使用计算节点

计算节点，顾名思义，即执行计算任务的节点，相比于数据节点，有以下2个核心特性：

• 不存储分布式表数据

  不存储分布式表数据、元数据文件，是一个轻状态的服务组件。计算节点的启停流程十分精简，任何故障都可以通过重启节点来进行快速修复。同时，计算节点可以进行快速地扩缩容。

  如下图所示 3 个数据节点、3 个计算节点的集群：

  
  

  图中 C1-CNODE1 至 C1-CNODE3 为计算节点，P1-DNODE1 至 P3-DNODE1 为数据节点。

  从文件系统来简单验证下计算节点的无状态特性。查看数据节点 volume 配置：

  # 查看存储卷
  select * from pnodeRun(getConfig{`volumes})
  P1-DNODE1    /ssd/ssd0/wfHuang/volumes/P1-DNODE1
  P2-DNODE1    /ssd/ssd0/wfHuang/volumes/P2-DNODE1
  P3-DNODE1    /ssd/ssd0/wfHuang/volumes/P3-DNODE1

  到操作系统去查看数据节点 P1-DNODE1 的 volume 目录：

  
  

  可看到 P1-DNODE1 作为数据节点，管理着事务 undo/redo 日志文件，和大量的数据块 CHUNKS 文件。但计算节点不需要配置 volume, 查看 C1-CNODE1 的默认 data 目录：

  
  

  可以看到计算节点 C1-CNODE1 不存储任何的分布式数据文件。

• 负责计算

  在分布式表读写场景中，由数据节点负责 map 阶段，计算节点负责 merge-reduce 阶段（详见通过一个-sql-查询理解计算节点）。在纯内存对象（常规内存表、分区内存表、流数据表等）计算场景中，由计算节点独立执行。

  执行一个分布式表查询 SQL，从文件层面来观察计算节点在数据查询发挥的作用。

  查看计算节点 C1-CNODE1 打开的文件（open files）：

  [appadmin@compute volumes]$ lsof -p pidOfComputenode
  skip library files
  compute_w 38563 appadmin   19u     IPv4 319181009       0t0       TCP compute:8961->192.192.168.2:56780 (ESTABLISHED)
  compute_w 38563 appadmin   20u     IPv4 319103550       0t0       TCP compute:8961->192.192.168.2:56786 (ESTABLISHED)
  compute_w 38563 appadmin   22u     IPv4 319095550       0t0       TCP compute:8961->192.192.168.2:34842 (ESTABLISHED)
  ...skip 20 lines......
  compute_w 38563 appadmin   44u     IPv4 319337513       0t0       TCP compute:42566->192.192.168.2:8961 (ESTABLISHED)

  查看数据节点 P1-DNODE1 打开的文件：

  [appadmin@support4 log]$ lsof -p pidOfDatanode
  compute_w 99763 appadmin   33u     IPv4 49451766       0t0       TCP support4:8961->192.192.168.1:42474 (ESTABLISHED)
  compute_w 99763 appadmin   36u     IPv4 49455618       0t0       TCP support4:8961->192.192.168.1:42562 (ESTABLISHED)
  compute_w 99763 appadmin   39r      REG     8,17 272358102 177313580 /ssd/ssd0/wfHuang/volumes/P1-DNODE1/CHUNKS/test_compute/20220102/hQE/machines_2/1_00000058
  compute_w 99763 appadmin   40r      REG     8,17 272358117 177313646 /ssd/ssd0/wfHuang/volumes/P1-DNODE1/CHUNKS/test_compute/20220106/hQE/machines_2/1_00000062
  compute_w 99763 appadmin   41r      REG     8,17 272358131 177313802 /ssd/ssd0/wfHuang/volumes/P1-DNODE1/CHUNKS/test_compute/20220227/hQE/machines_2/1_00000072
  compute_w 99763 appadmin   42r      REG     8,17 272358111 177313644 /ssd/ssd0/wfHuang/volumes/P1-DNODE1/CHUNKS/test_compute/20220109/hQE/machines_2/1_00000060
  ......skip hundreds of lines......

  可以观察到计算节点 C1-CNODE1 打开了大量的 TCP 连接符，而数据节点 P1-DNODE1 除了 TCP 连接符外，还打开了大量的磁盘文件。

  
  

  如上图所示，计算节点 computenode1 负责解析 client 的请求，并协调数据节点 datanode[1-3] 读/写数据（即 map，分布式执行过程）。接收到 datanode[1-3] 的数据后，经过 reduce-merge 过程，将最终结果返回给客户端。存算分离的架构，使得数据节点的物理资源得到隔离，写入和读取数据更为稳定、高效。

当客户端向计算节点发起一个 sql 查询：

pt1 = loadTable(database("dfs://szstock"),"sztrade")  
select [HINT_EXPLAIN]
    securityID, min(price) as LowPx, max(price) as HighPx, sum(tradeQty) as vol
    , first(price) as OpenPx, last(price) as LastPx, sum(tradeQty * price) as val
from pt
where execType="F"
group by tradedate,securityID,minute(temporalParse(tradetime$STRING, "yyyyMMddHHmmssSSS")) as minute
order by tradedate,securityID, minute

通过 [HINT_EXPLAIN] 获取执行计划，整个 sql 的执行过程依次为 parse，map，merge，reduce 四个部分。

{
    "measurement": "microsecond",
    "explain": {
        "from": {
            "cost": 2
        },
        "map": {
            "partitions": {
                "local": 0,
                "remote": 320
            },
            "cost": 5189927,
            "detail": {
...skip many lines...
        },
        "merge": {
            "cost": 446588,
            "rows": 8522241,
            "detail": {
 ...skip many lines...
                }
            }
        },
        "reduce": {
            "sql": "select [98307] * from 702a9832b17f0000 order by tradedate asc,securityID asc,minute asc",
            "explain": {
                "sort": {
                    "cost": 296783
                },
                "rows": 8522241,
                "cost": 746374
            }
        },
        "rows": 8522241,
        "cost": 6691112
    }
}

其中，query parse、merge、reduce 过程发生在计算节点，map 阶段则并行发生于各数据节点。

整个执行过程消耗资源具体如下表。

该 SQL 总共涉及 300 个分区，各分区记录数在 60-280 万不等。merge-reduce 阶段分布处理了 850 万的数据。由 C1-CNODE1 承担的 merge-reduce 阶段总耗时占比为

（446588 + 746374）/ 6691112 = 17.8%

上述例子 map 阶段 where，group by 算子大大减少了数据量，因此计算节点的算量不大，如果是读取数据类的场景，比如数据导入导出，此比例会大大增加。

执行过程中，计算节点与数据节点的内存消耗情况如下图所示（node33：计算节点，node34-36：数据节点，下同）。

可以看出在 map 阶段在数据节点，是分布式、并行执行的。在 merge-reduce 过程中计算节点的内存消耗逐渐增加。

在数据节点上执行复杂业务计算（如因子计算、机器学习等），会占用大量的内存。使得 cacheEngine 无法获取足够多的内存，而发生 cacheEngine 内存溢出等问题，导致写入延迟，甚至失败。

ChunkCacheEngine can't reclaim any memory after trying 13 times.

Transation failed, cacheEngine out of memory.

另外，使用 TSDB 引擎，数据刷盘时需进行排序，需要消耗大量的 CPU 资源以及排序 buffer。若 CPU 、内存资源不足时，TSDB 的写入性能会下降。

通过在架构上引入计算节点，采用计算与存储分离的架构设计，可以有效地保证集群数据写入的稳定性。

使用计算节点的集群能显著地降低故障平均修复时间，尽可能地减少对业务的影响。核心原因是由于计算节点不管理分布式数据，重启可以在数秒内完成。当某个计算节点出现拥塞、无法响应等情况，且无法热修复的，重启即可使其恢复正常服务。

如图，当数据节点启动时，流程中的蓝色标记部分：“回放 redo 日志”，“恢复损坏的 chunk” 等包含了大量的 IO 操作，实际生产环境中耗时为分钟级（具体视数据量大小而定）。而计算节点流程十分精简，可以在秒级完成启动。

在节点列表文件 cluster.nodes 中编写计算节点及代理节点。推荐计算节点以 CN (取 compute node 首字母) 标识，数据节点以 DN (取 datanode 首字母) 标识，如：

localSite,mode
192.193.168.2:8990:controller1,controller
192.193.168.3:8990:controller2,controller
192.193.168.4:8990:controller3,controller
192.193.168.2:8960:P1-agent,agent
192.193.168.3:8960:P2-agent,agent
192.193.168.4:8960:P3-agent,agent
192.193.168.1:8960:C1-agent,agent
192.193.168.2:8961:DN1,datanode
192.193.168.3:8961:DN2,datanode
192.193.168.4:8961:DN3,datanode
192.193.168.1:8961:CN1,computenode
192.193.168.1:8966:CN2,computenode
192.193.168.1:8967:CN3,computenode

在配置文件 cluster.cfg 中，需要使用前缀来区分数据节点与计算节点的配置项。

CN%.workerNum=64
CN%.maxConnections=3000
CN%.maxMemSize=256
CN%.dataSync=0
DN%.maxConnections=512
DN%.maxMemSize=128
DN%.workerNum=32
DN%.chunkCacheEngineMemSize=6
DN%.dataSync=1
DN%.redoLogDir=/ssd/ssd0/wfHuang/volumes/redoLog
DN%.volumes=/ssd/ssd0/wfHuang/volumes,/ssd/ssd1/wfHuang/volumes,/ssd/ssd2/wfHuang/volumes

CN% 匹配 CN 开头的计算节点，DN% 匹配 DN 开头的数据节点。以此保证计算节点与数据节点的配置项不会互相干扰。

注：计算节点必须配置 dataSync = 0 。

参考集群安装模式部署并启动 agent、controller 节点。启动完成后后，使用 web 启动计算节点。

计算节点监控的关键运行时参数主要是内存、连接数。如何避免个别 session 占用过多内存，而导致其他 session OOM，是运维管理上的重点。

1. 内存管理

   在内存紧张时，通过 getSessionMemory 视图进行会话管理，回收空闲会话：

   select 
       userId
       ,sessionId
       ,memSize/1024/1024 as MB
       ,localtime(createTime) as createTime
       ,localtime(lastActiveTime) as lastActiveTime
       ,(now() - localtime(lastActiveTime))/1000 as idleSeconds
   from getSessionMemoryStat()
   order by memSize desc

   通过 closeSessions(sessionId) 来关闭对应会话并回收内存。

   
   

   图示，test 用户已经空闲15分钟了，视情况可以关闭此 session，并回收1000MB的内存。

2. 连接管理

   在 Linux 中，DolphinDB 的一个 connection 对应一个 TCP 连接。在物联网场景下，需要调整连接数（每个 tcp 连接消耗 4K 内存，对操作系统消耗不大），以应对高 QPS 需求，建议将计算节点的配置设置为 maxConnections=5000 及以上。

   The number of incoming connections has reached the limit of 512

   当日志出现以上信息时，说明连接已经不足，检查程序是否存在连接泄露的情况。应急处理请使用 closeSessions() 清理一些空闲会话以回收连接，并增加 maxConnections 。

慢查询通常会引起系统雪崩，需要重点关注。DolphinDB 提供一个 job log 日志文件，记录 DFS 表相关查询日志，默认在 log 目录下，名称为 nodeAlias_job.log。

• 监控 top10 sql

  jobLogFile = getConfig(`jobLogFile)
  queryLog = loadText(jobLogFile)
  
  select 
      node 
      ,userId
      ,sessionId
      ,jobDesc as sqlText
      ,startTime
      ,endTime
      ,(endTime$DATETIME - startTime$DATETIME ) as costSeconds
  from queryLog
  order by (endTime$DATETIME - startTime$DATETIME ) desc
  limit 10

  
  

如图，根据 sqlText，userId，运行时间及时长，综合分析处置慢查询，比如限制一些用户进行较大规模的查询，以及对部分查询效率和速度较低的 SQL 语句进行优化。

1. 在新机器上部署和配置 agent

   拷贝原机器上的 agent 部署包到新机器，并修改 agent.cfg，如

   mode=agent
   workerNum=4
   localSite=192.192.168.4:8950:C4-agent
   controllerSite=192.192.168.4:8990:controller3
   sites=192.192.168.4:8950:C4-agent:agent,192.192.168.2:8990:controller1:controller,192.192.168.3:8990:controller2:controller,192.192.168.4:8990:controller3:controller
   lanCluster=0

   注意：主要修改新机器上代理节点的 IP 和端口，以及控制节点集群的 IP 和端口。

2. 在 web 集群管理界面修改 Nodes Setup 配置，新增 computenode 和 agent。

   
   

   注意：在添加完每个节点的信息之后，切记点一下该行最后的 +（加号），确保添加成功。

3. 在 server/ 目录下面执行如下命令，启动新节点的 agent

   nohup ./DolphinDB -console 0 -mode agent -home data -config config/agent.cfg -logFile log/agent.log &

4. 重启 controller 节点使配置生效

   回到 web 管理界面，可以看到已经新增了一个 agent4 并且是未启动状态。

   
   

   注意：代理节点的信息需要在重启集群的controller节点之后才能显示出来，数据节点和计算节点在添加完之后刷新页面就可以观察到。

   
   

5. 启动新计算节点

   
   

6. 进入新增计算机节点 web 页面，执行查询命令，检查新增数据节点是否能够正常工作

   
   

   注意：新计算节点命名应遵守部署文档中提到的命名规则，以确保 cluster.cfg 能赋予新的计算节点正确的配置。

1. 停止计算节点

   在 web 集群管理页面停止对应计算节点，等待其安全关闭，通常需要数秒。

2. 修改集群配置

   在 nodes Setup 中删除对应的 computenode 节点。

   注意：执行完上述操作之后，页面的计算节点信息并不会消失，需要重启集群的控制节点才会消失。