Apache IoTDB分布式是基于工业物联网领域存在的海量数据存储、高速数据读取和复杂数据分析需求进行设计的。为了有效地避免单个硬件、软件系统不可靠并且无法提供足够的计算和存储能力等问题，Apache IoTDB分布式采用了类P2P架构，支持集群的水平扩展，通过使用双层粒度数据管理方法等，Apache IoTDB能够高效率地利用异构集群中分布的计算和存储资源，在降低硬件投资的同时，保证存储、读取和分析的需求。

数据模型

Apache IoTDB的数据模型请参考。

集群逻辑单元

Apache IoTDB 分布式的逻辑结构图如下：

集群逻辑单元结构示例（5节点3副本，包含1个元数据组和5个数据组，每个数据组包含3个数据持有者）

一个完整的 Apache IoTDB 分布式是运行在一到多个集群节点上的，在逻辑上，它包含多个元数据持有者和数据分区持有者。系统中存在一个元数据组和节点数个数据组，每个集群节点均位于元数据组内和副本数个数据组内。应用可以通过提供的API接口与集群中任意一个节点进行互动。下面对上述概念进行简要的介绍：

• 集群节点（Node）：集群中的每一个节点在整个集群中不仅仅会扮演元数据持有者的角色，还会扮演一到多个不同数据分区持有者的角色。此外，每一个节点都可以作为协调者帮助用户完成相应的请求。

• 元数据组（Metadata Group）：元数据组存储了存储组元数据、分区映射关系、用户权限管理等集群元数据，其本身是一个Raft组，其中的每一个节点被称为元数据持有者。该组可以完成创建存储组等操作。

• 元数据持有者（Metadata holder）：元数据持有者属于元数据（Metadata）组。

• 每个元数据持有者都保留有全量的存储组信息，从而可以将查询请求根据存储组进行拆分。

• 以设置存储组请求为例，该请求会作为一条日志记录持久化到磁盘中，即元数据日志（Metadata log）。在上图中，每一个节点都位于Meta Group中，每个节点会包含本地状态机和元数据日志两部分。

• 本地状态机为Apache IoTDB 单机版系统，负责处理日志操作。

• 元数据日志中记录了设置存储组等操作。

• 数据组（Data Group）：数据组存储了一部分设备、时间序列元数据与数据等，其本身是一个Raft组，其中每个节点被称为数据分区持有者。该组可以完成创建/删除时间序列、写入/查询数据等操作。

• 数据分区持有者（Data partition holder）：数据分区持有者属于数据（Data）组。分布式中的数据按照<存储组, 时间段>为粒度进行数据分区，不同数据分区之间不存在数据重叠，一个数据分区中的数据存在于多个节点作为副本。

• 每个数据分区持有者持有自己的时间序列元数据，而需要其他元数据则需要从分布式其他节点上拉取，可以提高系统的效率和性能。

• 以创建时间序列为例，该请求会作为一条日志记录持久化到磁盘，即数据日志（Data log）。在上图中，每个节点都位于多个Data Group中，每个节点包含本地状态机和数据日志两部分。

• 本地状态机为Apache IoTDB单机系统，负责处理数据日志中的操作。

• 数据日志记录创建/删除时间序列和写入数据操作。

数据分区

时间序列的元数据存储在存储组的0号时间分区对应的数据组节点上，其他数据组如果想要使用对应时间序列的元数据则需要从对应存储组的0号时间分区中拉取元数据，并且将元数据在本地进行缓存。

IoTDB是通过存储组和时间段两个维度对时间序列数据进行划分，数据分区之间不存在数据交叠，因此数据分区的单位为<存储组，时间段>，以该粒度进行进行分区和调度。同一存储组的数据在磁盘上物理隔离，因此首先基于存储组进行数据的划分；为了防止某一数据分区过大，IoTDB对每个存储组基于相同的时间间隔再进行划分，这样同一存储组的数据被划分为多个数据分区，可交给不同的节点或副本组进行管理。

数据分配

IoTDB通过一致性哈希算法确定每个数据分区存储的位置。

数据复制

数据复制是将一份数据在多个物理地点上保存，这样即使一台机器出现宕机，其他的机器上的副本还能提供服务，用于提高系统的可用性；并且多个副本可以共同用于查询，提升数据查询的性能。IoTDB通过配置文件中的replication_num配置项设置副本数，在哈希环上顺时针选择副本数个节点作为一个数据副本组。

数据一致性

IoTDB选择Raft 算法用于维护多个副本间的数据一致性。Raft 是一致性协议，Raft的细节问题可以参考它的论文。

IoTDB中以Raft数据组（Data Group）进行数据复制的一致性维护，数据组内的所有节点维护同样的数据副本。每个数据组内的数据副本由该组所管理的哈希槽中的所有数据分区组成，每次数据写入都会在Raft组利用一条Raft log来进行同步，当超过Quorum数的组内成员收到log后，即可commit，提高了数据写入的效率。

数据写入

IoTDB集群遵循以下的写入流程：

• 用户选择集群中的任何一个节点作为协调者节点，建立连接并发送数据写入请求。

• 协调者节点先解析写入操作中的所有时间序列所属存储组，例如写入时间序列 "root.sg1.d1.s1" 和 "root.sg2.d1.s1" ，若存储组不存在时，会自动创建时间序列，并计算时间戳对应的时间片，然后根据一致性哈希算法计算所有被写入的时间序列所在的数据分区即 < 存储组，时间戳 > 对应的哈希槽，最后确定管理所有哈希槽的数据组，例如数据组1和数据组2。

• 协调者节点将写入请求转化的物理计划发送给数据组1和数据组2的任意节点。

• 在数据组1和数据组2的任意节点收到物理计划后，会转发给该数据组的leader，管理该数据的数据组leader分发日志并等到超过半数数据组成员收到后，数据组 leader 提交日志， 若该存储组内不存在被写入的时间序列，则会自动创建时间序列，之后再重新执行写入操作。

• 最后将结果返回给用户。

数据查询

• 对于数据查询请求，从整体而言，收到请求的协调者节点会首先解析该请求（可能需要从其他节点拉取时间序列元数据），接着会通过LocalSeriesReader / RemoteSeriesReader（是对Apache IoTDB 单机版SeriesReader的一个封装）的方式以batchData为单位从本地或远程来获取查询结果。

• Apache IoTDB 分布式查询的具体流程：解析SQL -> 逻辑计划 QueryOperator -> 物理计划 QueryPlan -> 物理计划执行Executor 生成结果集 -> 返回结果集。

• “解析SQL -> 逻辑计划 QueryOperator”：该步骤会在协调者节点上完成antlr解析和内部结构赋值等计算工作。

• “逻辑计划 QuryOperator -> 物理计划 QueryPlan”：该步骤涉及到与元数据的交互，包括类型获取/检查，通配符消除等等，往往需要协调者节点完成计算工作，也可能涉及到远程节点完成相应的IO操作。

• “物理计划 QueryPlan -> 执行物理计划”：

• 为每个待查序列创建SeriesReader：

• 对于本地序列，会建立LocalSeriesReader，并且保存在本地的DataSet中

• 对于远端序列，会创建RemoteSeriesReader，并且在远程注册一个queryId对应的LocalSeriesReader。

• 将所有待查序列看做本地序列，执行单机查询流程。

• “执行物理计划 -> 结果集返回”：

• 该步骤会不断的调用next函数来获取到数据：

• 对于本地的情况，会利用刚刚的LocalSeriesReader从本地的tsfile里面来获取BatchData。

• 对于远程的情况，会利用刚刚的RemoteSeriesReader通过RPC请求，利用远程节点的LocalSeriesReader从远程的TsFile里面获取BatchData。

• 在结果集返回后需要释放所有的资源：

• 对于本地的情况，直接释放对应的LocalSeriesReader的相关资源即可。

• 对于远程的情况，不仅仅需要释放本地RemoteSeriesReader相关的资源，还需要发送携带queryId的RPC请求释放对应远程节点的LocalSeriesReader的相关节点。

• 对于聚合，groupby 等可以将计算下推的查询，协调者节点会将该部分计算也下推到数据节点去执行，对应的数据节点只需返回计算结果个协调者节点即可。这样的设计可以更好地利用集群的计算资源。

• 查询一致性：提供强、中、弱三种级别的一致性，允许用户通过配置参数consistency_level来进行调整，主要针对的是数据组Follower read的场景，保证每个集群节点优先在本地进行查询，支持用户根据自己的具体业务场景来进行选择。

• 强一致性（strong consistency）：如果协调者节点注册的RemoteSeriesReader路由到follower节点，则follower节点需要先与leader节点进行通信并等待本地数据为最新数据后返回结果，如果和leader节点通信超时或失败则报错。

• 中一致性（mid consistency）：如果协调者节点注册的RemoteSeriesReader路由到follower节点，则follower节点需要先与leader节点进行通信并等待本地数据为最新数据后返回结果，不同的是如果和leader节点通信超时或失败则执行本地查询，并不会报错。

• 弱一致性（weak consistency）：如果协调者节点注册的RemoteSeriesReader路由到follower节点，则follower节点直接使用本地数据进行查询并返回节点，而不和leader节点进行同步。

• 中一致性和弱一致性满足最终一致性。

集群扩展

在面对以下的情况时，需要对集群进行扩展

• 集群所有节点负载都高，需要快速扩容。

• 集群内某几台节点负载很高，需要降低这些节点的压力。

在IoTDB的框架中，N个节点组成的集群，共有N个数据组，每个数据组存取一份包含多个数据分区的数据副本。在哈希环上，顺时针选取副本数个节点组成一个数据组。IoTDB 中基于一致性哈希的数据组分配方法在集群扩展时会造成多个数据组的成员替换。集群中包含多个数据组，每个数据组是由一致性哈希环中连续的副本数个节点构成。这种方式下在集群增删节点时不仅会存在数据组的增加或者删除，还会存在一些数据组进行了成员替换。 一致性哈希的存在，在集群的成员变更时，只会使得小部分数据进行在集群上进行迁移。

集群的成员变更必然会带来数据迁移，数据迁移的本质就是数据分区的重分配，哈希环上的哈希槽会在成员变更后，均匀分布到变更后的所有Raft数据组里。在IoTDB中，使用了哈希槽的概念，该方法将哈希环等分为 Q 个区间，每个区间称为槽，集群中的各个节点以槽为单位管理数据，为了减少数据的倾斜性槽数 Q 远大于集群中的节点数。假设集群节点数为 N，则各个节点管理的槽数为 Q/N，当集群删除节点时将负责的槽均匀地分配给集群中的其他节点；同理当新增节点时将均匀地从其他节点获得一定数量的槽。IoTDB中通过计算数据分区的哈希值找到对应的哈希槽，确定存储位置。

对于加入节点的成员变更，如下图所示，在 5 节点 3 副本的集群中加入节点 6 后除了增加一个数据组 6（数据组成员 {6，3，4}）外，数据组 1 和数据组 2 发生了成员替换。数据组 1 中新加入的节点 6 替换了节点 3，数据组成员由 {1，2，3} 变成了 {1，2，6}；数据组 2 中新加入的节点 6 替换了节点 4，数据组成员由 {2，3，4} 变成了 {2，6，3}。

对于删除节点的成员变更，在 5 节点 3 副本的集群中移除节点 3 后除了删除了一个数据组 3（数 据组成员 {3，4，5}）外，数据组 1 和数据组 2 发生了成员替换。数据组 1 中节点 4 替换了节点 3，数据组成员由 {1，2，3} 变成了 {1，2，4}；数据组 2 中节点 5 替换了节点 3，数据组成员由 {2，3，4} 变成了 {2，4，5}。

增加节点涉及原有 Raft 组的成员替换、并增加一个 Raft 数据组（直接新增 Raft 组并进行选举）。删除节点涉及原有Raft组的成员替换、并删除一个 Raft 数据组（等数据迁移完后进行删除）。Raft 组的成员替换不可以直接执行，否则可能会产生脑裂现象。为了解决这一问题，需要在原有 Raft 组内先增加新成员，再删除旧成员。

集群扩展时，会涉及多个 Raft 数据组的成员替换和一个 Raft 组的增删，整个过程分为两个阶段：

• 集群扩展第一阶段（集群整体行为）

• 元数据组 leader 进行安全性检查，确保之前的集群扩展已经完成。之后元数据组 leader 对数据进行重分区并将结果加入增删节点的日志中，然后将这条日志分发给所有元数据组的 follower，等到超过半数的 follower 收到这条日志。

• 数据组 leader 将这条日志发送给所有的 Raft 数据组。每个数据组收到日志后按照 Raft 流程进行处理，所有数据组成员在收到日志时直接提交日志，即更新本地的集群成员信息和数据分区信息并进行预成员变更（预成员变更是对发生成员替换的数据组执行增加节点的变更操作）。每个数据组 leader 按照预成员变更后的组内成员信息进行日志分发。

• 当所有的数据组增加成员完成后，第一阶段结束并返回给用户集群增加/删除节点成功的消息。若集群扩展操作为增加节点，新加入的节点会正式启动并提供服务。

• 集群扩展第二阶段（各节点内部行为）

• 各节点对本地所有数据组进行正式成员变更。

• 当由于成员替换导致本节点成为某个数据组的新成员时，本地创建并启动该数据组的实例提供服务；当由于成员替换导致某个数据组成员不包含本节点，本地关闭并删除该数据组的实例。

• 对所有的数据组按照数据重分区情况启动异步的数据迁移任务。当所有数据组的数据迁移完成后，用户就可以进行下一个集群扩展请求。

• 如果是删除节点操作，元数据组 leader 需要通知被删除的节点执行删除流程。 因为第二阶段时元数据组的成员已经不包含被删除的节点，不会和这个节点进行心跳，因此需要主动通知它执行删除流程。

客户端接口

• Apache IoTDB 分布式采用了P2P架构，集群中每一个节点都可以和客户端进行连接，具体连接方式和单机版相似，可以通过Java Session和JDBC等的方式来进行连接，对业务层透明。

• Apache IoTDB 分布式相对于单机版也提供了其他功能和性能上的优化，对于不同类型的请求，支持自动容错与重连尝试：

• 写入请求：

• 自动容错，减少业务层报错：分布式Session创建时允许用户传入多个节点的地址进行管理，提供自动容错和重连尝试的功能，避免向上层业务层报错。

• Cache leader优化：我们使用客户端（Java Session）缓存leader地址，从而使得客户端在写数据时可以避免进行多轮转发直接找到对应的正确节点，有效地降低了时延，减少了集群内部转发对性能的影响。

• 查询请求：对于查询请求，目前建议在业务层完成负载均衡，对不同的节点轮询查询请求来提高整个集群的吞吐。