Pinot 架构分析

Pinot的目标是在任意给定的数据集上提供分析。输入的数据可能存储于hadoop或kafka.在LinkedIn,大多数的跟踪数据被发布到kafka中并最终通过ETL处理转移至Hadoop.为了提供更快的分析，Pinot将数据组织为行模式并使用多种索引技术比如bitmap,翻转索引等。在Hadoop上的数据将通过map-reduce作业转换为索引分片。接着将索引分片提交至Pinot集群。Pinot服务者加载这些索引分片并为用户提供查询。数据的新鲜度取决于运行于Hadoop之上的那个map-reduce作业的频度。这通常需要持续几个小时。大多数应用对于数据的新鲜度和存在几个小时或几天的滞后是无感知的。但是，一些使用案例例如“谁看了我的简介”(WWMP)需要近实时的分析。为了支持这种使用案例，Pinot有能力在业务运行中直接从kafka中提取数据并生成索引分片。

数据流

存储至HDFS的数据可以为任意类型比如AVRO,CSV,JSON等。Pinot支持单表模式查询。如果查询的区域是分布式的跨越多个数据集合，客户端需要在生成Pinot索引分片之前将这些数据集合并至一个数据集。在这些索引在Hadoop上生成滞后，我们将这些数据推送至在线服务集群。Pinot服务者节点将加载这些索引并提供查询。索引分片将在预配置的保留时间之后将自动删除。

实时数据流与hadoop数据流稍有不同，实时数据流节点直接监听kafka数据流并在内存中生成索引分片并将其周期性的刷新至硬盘。查询通过内存中的实时数据并将分片持久化存储至硬盘。该想法是数据从kafka中直接推送至索引分片生成所在的hadoop并在保留期之前将数据推送至历史节点。

查询路由

从用户的角度看所有的请求都被送至Pinot 代理.用户不需要担心实时节点和历史节点查询。Pinot 代理足够智能能够分开处理实时节点和历史节点查询并在查询回应前合并结果集。例如，用户发出请求select count() from table where time > T 时间戳，Pinot理解这个时间是一个特定的维度并将该查询适当地分离为实时和历史查询。如果这个历史节点仍然包含昨天至今(T1)的数据,该代理将生成两份独立的查询#1 select count() from table where time > T and time < T1 #2. select count(*) from table where time > T1. 代理发送请求#1 至历史节点和#2至实时节点。选择历史集群为最多天数是为了获得更少的请求延迟。Hadoop中的数据通常以一天为单位进行汇总而实时集群中的数据仅仅以给定的分钟/小时时间段作为汇总。Hadoop允许我们在分片合成的时候使用先进的优化技术以提高查询性能。

Pinot组件结构

历史节点

数据准备

Pinot索引分片建立在Hadoop之上。客户端提供输入数据至Pinot分区创建的作业。Pinot期望该输入数据在需要使用之前预先合并。一般用户运行joins/decoration作业作为这一步的一部分。该数据可以为任意类型比如AVRO,CSV,JSON

在Hadoop之上建立分片

Pinot团队为生成分片提供了类库。Pinot期望数据存储于HDFS中。HDFS中的数据将被分割为256/512MB的小分片。Pinot索引生成作业将针对每个晓芬片生成一个映射并生成新的分片。在某些实例中，无关多少个分片和数据的大小，客户端需要固定数量的以每天为单位合成的分片。更多相关信息关于如何配置Hadoop管道来生成Pinot索引分片请参考Pinot 2.0Hadoop WorkFlow

从HDFS向NFS转移数据分片

Pinot分片在Hadoop上产生之后，其需要被传送至在线的服务集群。该任务通过Pinot团队提供的Hadoop服务者推送作业完成。该作业是通过基于azkaban的运行于Hadoop网关服务者上的非map-reduce的JAVA作业来实现。其从HDFS中读取数据文件并在Pinot控制器暴漏的任何一个终端上执行发送文件的指令HTTP POST来传输文件.该文件最终存储于挂载于控制器节点的NFS。

在存储分片于NFS之后，控制器将该分片分配至Pinot服务者中的任意一个。针对Pinot分配的分片将被Helix管理和维护。

从NFS向历史节点转移数据分片

分片对Pinot服务者的分配信息存储于Helix Idealstate中。Helix监控服务者的存活，当服务者启动时候，Helix将通知Pinot服务者关于这个分片的信息。该分片的元数据包含获取分片的URI信息并存储于以zoopkeeper为支撑的Helix属性存储中。Pinot服务者从控制器中下载分片文件并将其内容解压至本地磁盘。

分片加载

为压缩的分片包含针对每个列的前向索引和反向索引(如果被激活的)的元数据。基于加载模式(内存，mmap)索引分片即可载入内存或被服务者所映射。其作为OFFLINE-ONLENE过渡时期的一部分通过Helix触发。当该过渡成功地结束，该代理将通过Helix被通知到其已经准备好提供服务。Pinot 代理将启动针对拥有新分片的服务者的路由查询。

分片失效

Pinot中的大多数数据是基于时间分片并且保留策略是基于每个使用案例配置的(基准可以为天，月，年).Pinot控制器使用后台清理线程称为删除控制器线程来查看分片元数据并从NFS中删除失效的分片同时对应性的升级Helix中的IdealState. Helix通过追随一个OFFLINE-DROPPED的ONLINE-OFFLINE过渡时期来通知每个失效分片拥有者的失效分片删除操作。ONLINE-OFFLINE过渡时期将首先将该分片OFFLINE(已经不可查询)并且OFFLINE-DROPPED过渡时期将从本地磁盘删除该分片数据。

实时节点

追踪数据被发射到kafka中。数据被分割为分区并分布于多个kafka 代理中。当一个资源在Pinot中被建立，我们一个分配一组Pinot实例从一个指定的kafka话题中开始消费数据。在初始版本中，我们使用高等级的kafka消费者通过多个Pinot实例来分布消费。如果一个Pinot服务者当掉，该消费行为将在剩下的节点中重新分布。这种重新均衡机制在高等kafka消费者组中所控制。为了追求可靠性，Pinot可以基于每个kafka话题运行多于一个的消费者组。

分片创建

当pinot服务者消费一个预定义好的多个事件(比如500万个)，在内存中的数据将转换为一个离线分片(类似于hadoop作业)。当一个分片被成功生成，我们将在kafka中确认这个偏移量。如果这里失败，我们总师从最后的监测点来重启消费。分片生成的类型于hadoop作业生成的类型相同。这样做的原因是-这个新的生成分片可以被重分配为一个离线节点。其将变得非常方便当我们拥有该索引服务而且我们可以一天天的减少针对Hadoop管道的依赖。

分片失效

实时分片同时拥有基于使用案例需求而配置的数据保留机制。不同于离线分片其保留期可以为多个月，实时分片的保留期近可以为多天(通常为3天)。这样的原因是从kafka总消费数据而创建的实时分片无法使用我们可以针对hadoop而设定的所有优化。甚至，来自于kafka中的数据将在Hadoop上被终止并每日分片可能在Hadoop上被重新生成。其允许我们在分片创建过程中来应用高级优化技术例如更好的压缩技术，例如基于主key值的数据排序。另外值得声明的事情是kafka不确切的保证数据当接收后。其可能导致在结果中临时性的失去数据准确性(很不明显的)。但Hadoop允许我们来修复类似问题。

Pinot集群管理

所有的Pinot服务者和代理可以被Apache Helix所管理。Apache Helix是一个通用的分布式管理框架在一个分布式系统中管理分区和复制片。更多地信息请参考helix.apache.org。

Helix将节点基于他们的职能分割为三个逻辑模块：

1，参与者： 该节点实际拥有分布式资源 2，观众： 该节点仅简单的监测每个参与者的状态并根据性的路由请求。路由器，比如，需要知道该实例在哪个分区上驻留和他的状态以便于将请求路由到适当地终端上。 3，控制器：该节点将监测并控制这些参与者节点。他的职责是在一个集群中协调所有的过渡并在保证集群稳定性同时确保状态约束匹配。

Pinot术语和其对应的Helix的概念，更多地信息请参考 Pinot核心概念和术语。

• Pinot分片： 其标注为Helix分区。每个Pinot分区可以拥有多个拷贝简称复制品。

• Pinot表： 多个Pinot分区将包入一个逻辑实体简称Pinot表，从属于一个Pinot表的所有分区将拥有相同的架构。

• pinot服务者： 其标注为一个Helix参与者。Pinot服务者拥有该分片(Helix分区)属于一个或多个Pinot表(Helix资源)。

• Pinot 代理: 其标注为一个Helix 观众将监听集群的分片和Pinot服务的变化状态。为了在一个Pinot 代理中支持多租户，Pinot 代理同时被标注为Helix参与者。

  
  

Zookeeper: Zookeeper用来存储集群状态。其同时用来存储Helix和Pinot需要的配置信息。当动态的配置其指定为一个使用案例如表结构，多个分片和其他的元数据也存储于Zookeeper中。Zookeeper同时被Helix控制器用来与参与者和观众通信。Zookeeper为强一致性和错误冗余。我们一般在生产中运行3到5个Zookeeper。只有一个Zookeeper ensemble在所有Pinot集群中被共享使用。

Pinot控制器: 所有的管理命令比如创建分配的Pinot服务者和代理为每个使用案例，通过Pinot控制器创建新的表或更新新的分片。Pinot控制器将Helix控制器包在相同的进程中。所有的Pinot管理命令通过Helix管理APIs内部的转译为Helix命令。

• 分配Pinot服务者/代理：该命令运行于我们使用新的使用案例或针对存在的使用案例分配更多的资源。该参数简单的摄取使用案例#1来命名X #2.多个Pinot服务者S和多个代理 B需要该使用案例。Pinot控制器使用Helix标签API来标注一个集群中的S服务者实例和B 代理实例为X。其意味着属于使用案例X的所有随后的表将被分配给S Pinot实例的相同集合。B 代理将得到一个Helix状态过渡标注其已经被分配为为使用案例X服务所有的请求。所有针对该使用案例X的查询将经过这些代理。

• 新建表： 其将为一个表创建一个空的IdealState.该表也必须被标注为X其意味着这个表中的所有分片将被分配给拥有相同标签X的实例。附加的元数据例如表保留期，分配策略等都使用Helix属性存储Api存于Zookeeper中。

• 上传分片： Pinot控制器将一个分片实体添加至表IdeaState.多少个实体被添加是基于对于表T有多少个复制集被配置。其有可能让Helix通过使用AUTO Ideastate模式来决定对Pinot服务者实例分配多少个分片，当前的版本中我们使用CUSTOM Ideastate模式来实现。更多信息请参阅Helix Idealstate模式。Pinot控制器拥有其自有的分配策略。默认的分配策略是针对服务者其拥有最少的已经分配的分片来分配分片和它的复制品。其同事确保这些复制品未被分配给相同的节点。

Helix控制器： 正如以上章节所揭示所有的Pinot管理命令仅仅简单的被转换为Helix管理命令。Helix命令接着更新相关的存储于Zookeeper中的元数据。Helix控制器工作类似于整个系统的大脑和并所有的元数据更改转换为一系列的操作和负责在针对性的参与者上的所有该操作的执行。所有一切通过状态转换来实现。更多的信息请参照Helix 架构。Helix控制器同时负责监控Pinot服务者。如果一个Pinot服务者启动或关闭。Helix控制器检测到这些行为并针对性的更新外部视图。Pinot代理监测这些更改并动态的更新路由表(这个功能被Helix类库所提供)。所有的查询将基于该路由表被路由。

关于多租户在Pinot2.0版本中如何被解决的更多的信息请参考Pinot2.0中的多租户。

代理节点

代理的职责是将一个给定的查询路由到合适的Pinot服务者实例，收集相应的回复并合并回复至最终结果将其传送回客户端。其中的两个关键步骤为

服务发现：服务发现是一种机制来知晓哪个表被集群所拥有和这个表分片的位置以及每个分片的时间段。正如之前所解释的那样，其通过Helix类库来将存储于Zookeeper中的对应信息分割。代理使用这些信息不仅仅来计算这个节点的子集并针对性的传送请求。而且裁剪这些被查询的分片。该动作被查看查询中的时间段所实现并使用那个时间段来过滤基于其元数据中的这个时间段的值的分片。正因为相同的分片集中包含多个复制品，代理有相应的弹性来选择分片集中的任意一个来路由查询。Pinot通过选择服务者来执行多个策略。更多可用的方法为 #1.非标准的对所有节点分布这些分片。#2 贪婪算法： 增加/减少查询需要路由到的服务者的数量。 #3.针对每个分片的服务者随机选择。当#1和#2为一个给定的查询做出优化，他们可能对于所有的查询效率很差(在服务者中不均匀的加载)。可以说，局部最优不一定意味着全局最优。当前默认的算法为#3比如为每个分片随机选择服务者。

聚合通信：当代理完成计算出一组节点来路由该查询，请求将被路由至对应的Pinot服务者节点。每个服务者节点处理该请求并返回应答，代理将合并从每个服务者中返回的应答并回复给客户端。该合并逻辑依据于查询选用与Limit, aggregation, group by top K等。如果服务者中的任意一个未能处理该请求或超时。代理将返回部分结果。我们必须在功能中支持更多地模式，使代理能够针对该失败查询重新操作。另外的方式是总是将查询传递到多个服务者并使用最快反馈的那个服务者的返回结果。

Pinot索引分片

Pinot索引分片是一个裸数据的列式展现。裸数据通常被展示为一种面向行的格式其可以为AVRO,JSON,CSV等。转换面向行的格式为列式将减少存储空间ing允许针对特定的列的快速扫描。当查询为更新或在数据中读取某个行的时候面向行的格式是更加高效的。其通常为实例中使用OLTP当关系型数据库如Oracle,MySQL等被使用。当查询需要跨越很多行来产生结果时候，列式数据库在另一方面是高效的和更快的，其通常用于分析使用案例。当行的数量变为成千上万后速度的不同将更加明显。

列式同时提供了存储相关的优势。一些列包含了重复值.比如，如果该列是一类的国家以面向行的格式存储将需要相应的 varchar(100)*多行的空间。列式可以应用多种编码比如Fixed Bit和在此之上，更进一步使用压缩算法来压缩数据。这些技术也可以应用于面向行的存储，该编码和压缩可以更加高效当存储为列式。例如，在一个分片中的所有值都是相同的，我们所需要存储的是该值和其出现了多少次(RLE编码)。面向行的数据库无法充分利用该数据方式。 列式存储也有他们的劣势，创建高效的列式存储通常需要消耗更多时间并且一旦创建完成器将很难被更改。当是OLTP作业时其将是一个问题，通常上讲OLAP使用案例包含了时间序列的数据将是不可更改的。

索引分片分析

Pinot为了可管理性和高效性将整个数据分拆为多个分片。更多地信息请参考Pinot核心概念和术语。正如之前章节所描述的，行数据被分拆成列数据。本章节将介绍为每个可能的数据类型对应的列式数据格式。

分片实体

分片元数据(metadata.properties) 该文件包含了关于该分片的元数据比如：

属性名称 描述

segment.name 分片名称 如cars_daily_2015-04-11_2015-04-11_1

segment.resource.name 资源名称 如cars

segment.table.name 表名

segment.dimension.column.names 多维的列名称 如模式，标签等

segment.metric.column.names 度量 列名称 如价格，多少

segment.time.column.name 列表示时间 如年

segment.time.interval 分片中的时间区间 可以为每小时，每周，每月等

segment.total.docs 10000

segment.start.time 分片关联数据中的最小时间值

segment.end.time 分片关联数据中的最小时间值

segment.time.unit 起始和结束时间的时间单元

元数据中的部分值将在查询处理过程中被使用。比如，一个分片可以被裁减如果该查询时间段与分片的min-max时间段不能产生重叠。 除了上面分片相关的元数据之外，我们也为每列存储了相应的元数据。以下描述了该属性其存储在一个基于列的基础之上。

列属性名 描述

column..cardinality 分片中对于该列的多少个唯一值

column..totalDocs 分片中多少个文档

column..dataType 列的数据类型 如INT, FLOAT,STRINGD等

column..bitsPerElement 如果字典编码被应用，需要多少个字节来存储每个值 column..lengthOfEachEntry 如果列类型为字符，其声明多长的字符需要来存储该值。类似于varchar(100).

column..columnType 列类型 维度，度量，时间

column..isSorted 在分片中列的值是否为排序的

column..hasValue 该列是否可以为空值

column..hasDictionary 是否一个字典被用来针对数据编码

column..hasInvertedIndex 是否该列有反向索引

column..isSingleValues 该列为单值或多值的模式

column..maxNumberOfMultiValues 可分配为多值

column.Max 每个文档有多值

column..totalNumberOfEntries 可分配多列，所有的多个实体跨越分片中的所有文档。

创建元数据(creation.meta)字典(.dict) 在列中字典用来编码所有的值。应用字典编码可以大幅度的减少数据大小。特别在例子中当列的基数是低的(最多几千)。一个选项是总是使用整数数据类型来编码改值。但在某些例子中唯一值的数量很少从而我们需要使用固定比特编码。例如，如果多个唯一值是3，我们仅仅需要2个比特来表示该实际值。字典文件存储了字典ID和实际值之间的映射关系。 字典编码节省了很多空间，其在查询处理中增加了更多的查找成本。我们需要实时的转换字典ID为实际值。我们都知道当查找需求很高时候会造成了显著地开销。查找无法被哈希映射因为哈希映射需要更多地内存。通过字典之上的线性扫描来实现一个查询将会非常慢，一个简单的优化是将值排序并为查找执行一个二分查找。基于字典的这些属性。我们无法盲目地在所有的列上应用字典编码。需要说明的是，我们已经在Pinot2.0当前的版本中为所有的列使用了字典。甚至在未来我们将应用字典编码当技术是<10k并仅为多维列。 字典不允许我们来加快查询处理。因为字典仅允许我们来得知一个分片中的一个列的唯一值。我们在谓词评估中可以跳过针对一个分片的处理如果在该字典中的谓词的RHS不存在。 当下我们基于每个分片产生一个字典，将来我们将探索思路通过在所有分片中维护一个全局的字典。其将进一步减少字典开销并允许我们将字典之上的查找转换为哈希图上的查找。

前向索引(.sv.sorted.fwd) 前向索引为一个给定的文档存储了列值(单个或多个)。前向索引被存储为一种格式器允许基于给定的一个文档ID的固定时间查找。在查找处理过程中，过滤词将返回一批文档IDs.前向索引将被使用为了获取一个指定的文档对应的行值。 单值排序性前向索引(.sv.sorted.fwd) 我们知道在任意行上排序行值将大幅度加速查询执行。大多数使用案例选择其中的一个列作为他们的主键。在索引创建中，Pinot在每个分片中基于该主键进行数据排序。排序数据允许我们大幅度的减少列的大小。比如一个分片中有1亿行而唯一索引仅有1百万。我们可以简单的为每一个唯一主键存储开始和结束doc id而不需要为1亿值排序。 单值非排序性前向索引(.sv.unsorted.fwd) 如果一列中的值未被排序。我们有以下的可能优化手段 1.尽可能的使用字典编码。请参考我们应用字典编码的章节 2.Pinot当前版本的Snappy或LZ4或ZLIB压缩，我们仅使用字典压缩器允许我们使用固定比特编码来压缩数据。接下来的版本，我们将评估其他压缩技术比如snappy等。这些压缩技术确实节省空间但需要更多地开销来解压他们在运行中。这里的挑战是在数据压缩和查询延迟间得到正确的权衡。 多值前向索引(.sv.sorted.fwd) 在某些例子中列式多值的比如成员的技能。当字典编码可以应用于多值列，前向索引可以如同单值使用案例一样简单因为每个列中的多个值可以被抽象出来。这里的挑战性是如何抽取一个给定文档id的值而不需要扫描整个数据集。为了实现该功能，我们创建了附加的头分区来线性的方式存储起始，结束偏移量。下面的图片将详细描述该实现。头分区需要更多的存储空间。如果该开销变得非常巨大。我们将重新排序来跳过基于列表的实现其类似于之前的想法但我们为每个N文档(桶)维护起始，结束偏移量。而不是为每个文档ID维护起始，结束偏移量。提取一个值需要我们首先定位该桶然后执行一个桶内部的线性扫描来定位值。附加的bitmap被维护用来标注该文档ID的起始偏移量和结束偏移。

查询处理过程

查询执行阶段

• 查询分析：Pinot支持一个简单定制版本的SQL我们声明为PQL。PQL仅仅支持SQL的一个子集。例如Pinot不支持Joins,嵌入式子查询等。我们使用Antlr来分析该查询到一个分析树。在该阶段，所有的语法验证被执行并且默认值被设定为缺失的元素。

• 逻辑规划阶段：该阶段在查询分析阶段中使用并输出一个逻辑计划树。该阶段为单线程并且非常简单和构建基于查询类型(selection,aggregation,group by等)的合适的逻辑规划算子树和被数据源所提供的元数据。

• 物理规划阶段： 该阶段进一步优化基于独立分片的规划。该优化应用于此阶段可以根据多种分片的不同而不同。

• 执行服务：当我们拥有每个分片的物理算子树，执行服务将接收负责在每个分片上的查询处理任务的调度。

－－－－－－－－－－－－－－－－－－

关注如下我的微信公众号“董老师在硅谷”，关注硅谷趋势，一起学习成长。