Cluster Mode

In cluster mode, DolphinDB configuration files are saved under the config directory.

  • controller.cfg: the configuration file for controller nodes

  • agent.cfg: the configuration file for agent nodes

  • cluster.cfg: the configuration file for data nodes

  • cluster.nodes: network information and node mode of each node in the cluster.

Note: All configuration files require that the first line is not a blank line.

For a high-availability cluster, all the configuration files cluster.cfg and cluster.node are synchronously managed by the raft group. After the configuration parameters are modified through the web interface, the interface will automatically synchronize the parameters to the configuration files in the cluster.

Controller

Configuration parameters for controllers can be specified in controller.cfg, or in command line when starting a controller node. localSite and mode must be specified.

Parameter Description
localSite IP address, port number and alias of the controller node.
mode=controller Node mode. Must specify "controller" here for a controller.
clusterConfig=cluster.cfg Configuration file for the cluster. The default value is cluster.cfg. It can only be specified in command line.
nodesFile=cluster.nodes Specify IP address, port number, alias and node mode of all nodes in a cluster. It is loaded when the controller is started. The default value is cluster.nodes. It can only be specified in command line.
dataSync=0 Whether database logs are forced to persist to disk before the transaction is committed. If dataSync=1, the database log (including redo log, undo log, edit log of data nodes,and edit log of controller nodes) must be written to disk before each transaction is committed. This guarantees that data written to the database is not lost in case of systemcrash or power outage. If dataSync=0 (default), the log files are written to cache before a transaction is committed. The operating system will write the log files to disk at a later time. We may experience loss of data or corruption of the database in case of system crash or power outage.
dfsMetaDir =/home/DolphinDB/server The directory to save the metadata of the distributed file system on the controller node. The default value is the DolphinDB home directory specified by parameter home.
dfsMetaLogFilename =DFSMetaLog The edit log file for metadata of the distributed file system on the controller node. The default value is DFSMetaLog.
PublicName Internet IP or domain name of the controller node. It must be a domain name if enableHttps=true.
datanodeRestartInterval=0 It is a non-negative integer of int type. Its unit is second. Defaults to 0, indicating that the system will not automatically start data/compute nodes. Configure this parameter to implement the following features: (1) Automatically start the data/compute node after the controller is started (the agent needs to be started); (2) The data/compute node will be automatically restarted after its offline time is larger than the set value (for a non-graceful shutdown and the agent needs to be started).
datanodeRestartLevel=CRASH Specifies the triggering condition for the controller to automatically restart the data/compute node. It can take the following values (case sensitive):
  • CRASH (default): The controller will automatically restart a node if the node was not terminated gracefully and did not send heartbeats after datanodeRestartInterval.
  • OFFLINE: The controller will automatically restart a node if the node did not send heartbeats after datanodeRestartInterval.
dfsRecoveryConcurrency The number of concurrent recovery tasks in a node recovery. The default value is twice the number of all nodes except the agent.
dfsRebalanceConcurrency The number of concurrent rebalance tasks in a node rebalance. The default value is twice the number of all nodes except the agent.
dfsChunkNodeHeartBeatTimeout

An integer (of INT type) indicating the time interval (in seconds) for marking a

data node/compute node/agent as down, i.e., if the controller has not received heartbeat message from a node for more than this time interval, the node will be marked and treated as down by default. The default value is 8.

The following configuration parameters are for high availability and are specified in controller.cfg.

Parameter Description
dfsHAMode=Raft Whether the controller nodes form a raft group. It can take the value of either Raft or None. The default value is None.
dfsReplicationFactor=2 The number of all replicas for each data chunk. The default value is 2.
dfsReplicaReliabilityLevel=0 Whether multiple replicas can reside on the same node.
  • 0 (default): Yes.
  • 1: No.
  • 2: replicas are allocated to multiple servers if possible.
dfsRecoveryWaitTime=0 Length of time (in milliseconds) the controller waits after a data node becomes unavailable before restoring the data to other data nodes. The default value is 0indicating no automatic recovery.
raftElectionTick=800 Specify a time interval (in 10ms): [raftElectionTick, 2*raftElectionTick]. After receiving the last heartbeat from the leader, if a follower does not receive the next heartbeat after a random waiting time within the specified interval, it will send a request for leader election. The default value is 800 (8 seconds) and the interval is [8s, 16s]. Note: All controllers of the raft group must have the same configurations.
Note:

The number of online data nodes must be no smaller than the value of dfsReplicationFactor, otherwise an exception will be thrown.

Agent

Configuration parameters for agents can be specified in agent.cfg, or in command line when starting an agent node. All of the following 3 parameters must be specified.

Parameter Description
localSite Host address, port number and alias of the local node.
mode=agent Node mode. Must specify "agent" here for an agent.
controllerSite Host address, port number and alias of the controller site of the agent node. When an agent starts up, it uses this information to contact the controller. It must be identical as localSite of one of the controllers in controller.cfg.
lanCluster=1 Whether the cluster is within a LAN (local area network). lanCluster = 1: use UDP for heartbeats; lanCluster = 0: use TCP for heartbeats. Set lanCluster = 0 if the system is deployed in the cloud.

Data Node

All configuration parameters in the standalone mode can be used for data nodes in the cluster mode. They are specified in cluster.cfg. Please refer to Standalone Mode for more details.

The following table lists the configuration parameters that apply to cluster mode only.

Parameter Description
lanCluster=true Whether the cluster is within a LAN (local area network). lanCluster = true (default): use UDP for heartbeats; lanCluster = false: use TCP for heartbeats. Set lanCluster = false if the system is deployed in the cloud.
allowVolumeCreation=1 Set the parameter to determine whether a volume can be created automatically if the volume does not exist. The value can be 0 or 1.The default value 1 indicates to create volumes automatically. If the parameter is set to 0, the system will report an error if the volume does not exist.
volumeUsageThreshold=0.97 Specifies the upper limit of volume usage on a data node. It is a floating-point value in (0, 1]. The default value is 0.97. When the volume usage on a data node reaches the threshold, new chunks are not allowed to be created on the node, but data can be written to existing chunks. The parameter can only be configured on the controller.
volumes=/hdd/hdd1/volumes, /hdd/hdd2/volumes, /hdd/hdd3/volumes, /hdd/hdd4/volumes The directory of data files. If it is not specified, the default directory is <HomeDir>/<nodeAlias>/storage.
recoveryWorkers = 1 The number of workers that can be used to recover chunks synchronously in node recovery. The default value is 1.
memLimitOfQueryResult The memory limit for a query result. The default value is min(50% * maxMemSize, 8G), and it must be smaller than 80% * maxMemSize.
memLimitOfTaskGroupResult In the Map phase of MapReduce, a single query is divided into several tasks, among which the remote tasks are sent to remote nodes in batches (task groups). The configuration parameter is used to set the memory limit of a task group sent from the current node. The default value is min(20% * maxMemSize, 2G),and it must be smaller than 50% * maxMemSize.

memLimitOfAllTempResults

Specifies the total memory limit (in GB) for all temporary results generated during SQL distributed queries. The default value is 20% * maxMemSize. If the memory usage exceeds the limit, temporary results are spilled to disk.

Streaming-related configuration parameters are specified in cluster.cfg.

Parameter Description
streamingHADir=<HomeDir>/log/streamLog The directory to keep streaming raft log files. The default value is <HomeDir>/log/streamLog. Each data node should be configured with different streamingHADir.
streamingHAMode=raft Enable high-availability for streaming.
streamingRaftGroups = 2:NODE1:NODE2:NODE3,3:NODE3:NODE4:NODE5 Information about raft groups. Each raft group is represented by group ID and aliases of data nodes in the group, separated with colon (:). Raft group ID must be an integer greater than 1. Each raft group has at least 3 data nodes. Use comma (,) to seperate multiple raft groups.

Asynchronous Replication

  • Master Cluster

    The parameters clusterReplicationSlaveNum and clusterReplicationMode must be configured.

    Parameter Description Node
    clusterReplicationSlaveNum Specifies the upper limit of slave clusters. The default value is 2. controller
    clusterReplicationMode Specifies whether the cluster is master or slave cluster. data node
    clusterReplicationWorkDir Specifies the working directory where data of write tasks on the master cluster is stored. The default value is <HomeDir>/clusterReplication. A larger-sized SSD is recommended. data node
    clusterReplicationSyncPersistence Specifies a Boolean value indicating whether to persist data of write tasks synchronously or asynchronously. Note: Synchronous persistence ensures data consistency but may affect transaction efficiency, while asynchronous persistence improves performance but may cause data loss in node crashes. data node

  • Slave cluster

    The parameters clusterReplicationMasterCtl and clusterReplicationMode must be configured.

    Parameter Description Node
    clusterReplicationMasterCtl Specifies ip:port of the controller in the master cluster. For a high-availability (HA) cluster, you can specify any controller in a raft group. controller
    clusterReplicationMode Specifies whether the cluster is master or slave cluster. data node
    clusterReplicationExecutionUsername Specifies the username (default=admin) used to execute operations regarding cluster replication. The specified user must have relevant privileges on transactions, otherwise the asynchronous tasks will fail. data node
    clusterReplicationExecutionPassword Specifies the user password (default=123456) for clusterReplicationExecutionUsername.
    Note: since version 2.00.13/3.00.1, the user is not required to specify this parameter.
    		 data node
    clusterReplicationQueue A positive integer specifying the number of execution queues. The default value is 4 times the number of data nodes. controller
    clusterReplicationWorkerNum The number of workers (task execution threads) on each data node. The default value is max(workerNum/4, 4). data node

Tiered Storage

For more information, see Tiered Storage.

To enable tiered storage, configure the storage path for cold data first.

Parameter Description
coldVolumes= [file://home/mypath/hdd](file://home/mypath/hdd), s3://bucket1/data Configures the storage path of cold data. After the tiered storage is enabled with functions moveHotDataToColdVolume and setRetentionPolicy, cold data will be moved from volumes to coldVolumes.

To store expired data in the cloud, DolphinDB provides the following AWS configurations.

Parameters Description
s3AccessKeyId ID of an Amazon S3 account
s3SecretAccessKey Secret access key of an Amazon S3 account
s3Region Region of an Amazon S3 bucket
s3Endpoint Endpoint to access Amazon S3
Note:

  • When configuring s3Endpoint, you can add a BOOLEAN value after the endpoint to indicate whether to access the endpoint via HTTP protocol (true) or HTTPS protocol (false, default). For example, s3Endpoint=192.168.1.160:980,true indicates accessing 192.168.1.160:980 via HTTP protocol.

  • With s3Endpoint configured, it is not necessary to configure s3Region.

Parameter Specification

In cluster.cfg, we can specify configuration parameter values in the following 4 ways:

  1. Node alias qualified configuration parameters. Node aliases are defined in cluster.nodes.

    nodeA.volumes = /DFSRoot/nodeA
nodeB.volumes = /DFSRoot/nodeB

  2. (Node alias + wildcard character ("?" or "%")) qualified configuration parameters. "?" represents a single character; "%" represents 0, 1 or multiple characters.

    %8821.volumes = /DFSRoot/data8821
%8822.volumes = /DFSRoot/data8822
DFSnode?.maxMemSize=16

  3. Use macro variable <ALIAS> for assignments with node aliases. For a cluster with 2 data nodes nodeA and nodeB:

    volumes = /DFSRoot/<ALIAS>

    which is equivalent to

    nodeA.volumes = /DFSRoot/nodeA
nodeB.volumes = /DFSRoot/nodeB

    Note: The macro variable <ALIAS> cannot be used to specify the volume for a single node, otherwise the controller cannot be started. For example, do not specify configuration parameter as nodeA.volumes = /DFSRoot/<ALIAS>.

  4. If the parameters are not qualified by node aliases, they indicate common configuration parameter values for all the data nodes in the cluster.

    // for Windows
maxConnections=64
// for Linux
maxConnections=512

maxMemSize=12

Parameter Assignment Order

A configuration parameter may appear in the command line or multiple configuration files. It may be assigned different values in different locations. DolphinDB checks for parameter values in the order below. If a parameter is assigned value in a step, then assignments in all subsequent steps are ignored.

  1. command line

  2. cluster.nodes

  3. qualified by specific node aliases in cluster.cfg

  4. qualified by node aliases and wildcard characters in cluster.cfg

  5. common configuration in cluster.cfg

Inter-node Communication

The following TCP configuration parameters allows users to optimize communication. Enabling these parameter can reduce transmission latency, but may increase network load.
Parameter Description
tcpNoDelay=1 Whether to enable the TCP_NODELAY socket option. The default value is false.
tcpUserTimeout=300000 Set the socket option TCP_USER_TIMEOUT. The default value is 300000 (in ms).
The controller monitors the status of other nodes through a heartbeat mechanism, which can use either TCP or UDP for transmission. The related configuration parameter are as follows:
Parameter Description
lanCluster=true Whether the cluster is within a LAN (local area network). lanCluster = true (default): use UDP for heartbeats; lanCluster = false: use TCP for heartbeats. Set lanCluster = false if the system is deployed in the cloud.
dfsChunkNodeHeartBeatTimeout

An integer (of INT type) indicating the time interval (in seconds) for marking a

data node/compute node/agent as down, i.e., if the controller has not received heartbeat message from a node for more than this time interval, the node will be marked and treated as down by default. The default value is 8.
The maximum number of connections on a node is controlled by the following configuration parameter:
Parameter Description
maxConnectionPerSite=2 The maximum number of connections from the local node to remote nodes.