GreenPlum

基于PostgreSQL大型并行处理(Massively parallel processing, MPP, 即share-nothing architecture)数据库

Greenplum DB 布署图

Master : 主机

Greenplum DB的入口, 安装有PostgreSQL 8.2.10,使客户端可以用psql连接连入或用API如JDBC,ODBC. 存放global system catalog.(系统表的集合,包含有Greenplum DB的元数据.但不包括用户数据.

Segment : 段

用于存放数据(用户自定义表和索引)和处理查询的主要地方. Segment实例是数据库服务器进程.推荐用一个有效的CPU或CPU核来服务一个Segment.

Interconnect : 互联

Greenplum DB的网络层面.采用标准的标准的千兆以太网.默认使用UDP,如使用TCP会有1000个段的限制

Greenplum DB可选的支持冗余和故障恢复,无SPOF.有如下保障措施

1) 段监控:

可设置镜像段,,可以用镜像段在主段失效的时候恢复.条件是要有足够的机器

在镜像打开的时候,系统会自动故障恢复,当Master连接一个段失败,它会标记这个段是不可用的直到这个段重新上线,Master会重标记它为可用.

故障恢复要求系统配置成fault operational mode.在只读模式下,DDL,DML命令瘵不能执行.而continue mode,每个数据块只要有有一个段实例在运行,系统会连续运行,一旦有一个失败,系统必须关闭来恢复失效的段. 如果镜像没有打开.

2) 主机监控:

可以布署主机实例的镜像或备份.备份主机采用warm standby(热备)方式.备份主机通过复制transaction log的方法同步.

3) 冗余的互联:

双千兆交换机,和冗余的连线.

并行数据装载:

外部表装载.如果采用Greenplum DB并行文件服务器(gpfdist),可以达到装载效率2T/小时

Greenplum如何存贮数据:

采用star schema的一个例子, factor table是sale, dimension table是customer, vendor, product.

如上图所示,sale, customer, product, vendor分别是不同的表.

GreenplumDB的分布策略:

1) Hash分布 : 这是默认的,一个或多个表的列的值做为分布键(distribution key),这个分布的Key用来hash每一行到一个特定段.相同的键值,总是被Hash到同一个段.如果没有定义,一般以主键或第一个列作为分布键

2)随机分布:用round-robin方式分布每行记录.这种性能不如Hash分布好

查询处理:

当Greenplum从前端接收到SQL语句,主机会分析这个查询,并做优化,创建并发查询计划(parallel query plan),主机分发这个计划给所有的segment来执行.每个Segment执行本地数据库操作来响应.

除了典型的数据库操作(扫描,联接)GreenplumDB还有附加的操作类型称为motion,一个motion操作有在段之间移动数据元组的动作.

GreenplumDB将工作从查询计划切分为slices.一个分片是能在段的层次上独立完成的一部分工作(不单是一个工作项,而有可能是多个工作项).当一个motion发生时,在参与段的两端各有一个slice.

GreenplumDB创建一批量的进程来处理查询的工作.在主机上查询的工作进程称为query dispatcher (QD), QD负责创建分发查询计划,并汇总和展现最终结果.在段上,一个查询工作进程称为query executor(QE). QE负责完成工作的片断.每个slice都有被指派到一个工作进程上. 每个段都有一定量的进程进行并发的服务.相关的工作于同一slice查询的进程被称为gang.

MapReduce:

允许开发者写Map和Reduce用多种的脚本语言,如Python和Perl.GreenPlum的MapReduce提供开发者访问数据的弹性,可以在文件中,可以在网站上,甚至是任意的操作系统命令.Greenplum提供这些数据相对于传统数据库而言无额外的开销:无锁,日志,分布提交协议.在另一方面,对必须要由RDBMS全功能保护的数据,Greenplum MapReduce 提供了有效的原生方法访问数据记录.它会把MapReduce程序推向Greenplum并行的数据库引擎上,来获取数据.

MapReduce的特性:

1) 由于GreenplumDB表分布在多台机器上,初始的Map阶段运行在数据分区所在的数据引擎上.

2) Greenplum扩展表机制允许文件和数据生成程序被注册为只读表,然后通过SQL语句来查询.

3) Greenplum能存贮MapReduce的结果到一个文件系统,但能同样方便的保存到GreenplumDB,并且有全的ACID支持及可选的子查询通过BI工具分析这些输出.产生结果也是并发的

4) Greenplum MapReduce framework能使用任意的SQL查询作为输入.另一方面,Greenplum MapReduce脚本能被注册成数据库中的视图(view)来作为查询中的虚拟表.

报价:

Perpetual license : $ 16K / core, 或者说 $70K/terabyte. 年维护费22%维护价. 或50%永久许可证下年订阅费. 可以再商量

Infobright

ICE (InfoBright Community Edition)

Features:

• Data : SQL-92,加SQL-99的扩展,对视图和存贮过程的全支持
• 灵活的模式支持
• 业界标准的接口 : ODBC,JDBC,及原生的连接方式
• API : C/C++,C#, Java(及原生的java驱动实现),Perl, PHP,Python,Ruby...
• 并发的用户: 500个用户,32个并发查询
• 操作系统: WindowsXP(32-bits), RHEL5(64b), Debian(64b), CentOS5(64b), Fedora9(32b), Ubuntu8.04(32b)
• 处理器:AMD 和Intel x86, 推荐16G RAM
• 高性能的数据装载器：一个定制装载器,跨多表的并发处理

好处:

• 小的数据尺寸: ICE的压缩率为10:1
• 高的扩展能力: 到50TB,(压缩后5T)单个服务器
• Column-Based
• 采用现成的硬件
• 简单的管理: 利用MySQL的管理工具,并包括在ICE中了.
• 低维护:自动管理知识节点
• 装载即运行(Load and Go) : 用你现有的Schema不要物理视图,数据分区和索引的实现.
• 支持装载格式: 可以装载文本文件
• BI灵活:与MySQL集成,可以通过它受BI工具的支持,如Cognos, Business Object, SAS, Pentaho, JasperSoft, MicroStrategy.

IB的Architectrue的好处:

• 理想的数据卷大小到30TB
• 市场上领先的数据压缩技术(10:1到超过40:1),极大的减小了IO及存贮
• 无许可证费
• 对复杂分析查询的响应
• 查询和装载时间在DB增长时保持不变
• 不需要特别的Schema
• 不需要物理视图,分区策略,或索引
• 简单的建立和管理
• 减少DW(data warehouse的成本)通过减少服务器的数量,还有总控的花费
• 运行在低成本的现成的硬件上
• 兼容BI工具 : Actuate, ...

IB的Architecture

• Column Orientation
• Data packs和Data pack 节点

三个层面:

Data Packs : 一个列中65536个项目被组在一起称为Data Packs. Data Packs改善了压缩率,因为可以根据数据类型提供压缩算法

Data Pack Node(DPNs) : 包含一个静态统计信息针对一个DP,DPN总是存在,这区别于传统数据库的索引

Knownledge Node (知识节点),是一个元数据集有关于DP或列之间的关系. 描述值的发生范围,也可描述与其它DB中数据的关系. 用于复杂的多表查询,如表联接,子查询(sub-queries)大多数都是在装载时创建,也有在响应时创建来优化性能.

• 知识节点和知识网格

DPN与KN组了Knowledge Grid. 与索引不同的是它们不是手工产生的,而是自动生成,从根本上说,它们建立了DB高层次的视图

• 优化器

如何不用索引进行复杂的分析查询

1) 利用DPNs(一些统计信息),如对于整型的列数据来说,一个DP有64k个数据项,对应的DPN就记录了这些数据项的最大值,最小值,及总和.有了这些数据在查询中,如果用对这列比大小的条件查询就可以区分出三类的DP

i) 相关的DP, DP内的数据全部落在条件以内,(因此要对这个DP解压缩数据)

ii)不相关的DP, DP内的数据全部落在条件以外(,因此可以忽略这个DP)

iii)不确定的DP, DP内的数据部分落在条件以内(,因此要对这个DP解压缩)

分析出三类DP后可以对特定的DP解压缩再分检其中的数据

DPN可以针对不同数据类型,提供一些扩展的统计信息,如对于字母数字的数据类型,提供特定的字符在特定的位置出现的次数.

左边是一个DPN的例子,说明一个表T有两列A,B,共有64k * 5条以内的记录

也就是对于A,B各有5个DP及对应的DPN.对于DP A1(A列的第一个Pack, 共64k记录),DPN中记下它其中的最小值是0,最大值是5,总合(SUM)是10000.

KN的例子是如有如下查询

SELECT MAX(X.D) FROM T JOIN X ON T.B = X.C WHERE T.A > 6;

其中T即上述的T表, X是另一个表,有两个整型列C,D,共64k*3以内的记录,也就是说有DP C1~C3, D1~D3.

对其中条件T.B=X.C,可以利用DPN生成如下结果,称为KN

这说明了一个关系,T.B和X.C有相等可能两个DP之间是标的1,否则是0.这是通过比较DPN中最大值最小值得出的结论.

• 数据压缩

用了空值Mask,(估计)有一个8k的位图,来表示某个DP中的值是空的.InfoBrigth有一套数据类型对应的压缩算法(专利申请中)

InfoBright如何利用MySQL

InfoBright实现了MySQL中的数据存贮引擎,如(MyISAM,InnoDB)等.

InfoBright采用的是Rough Set Mathematics做为基础

2000年时一个波兰数学家的团队开始改进Rough Set理论,研究出一个既有精度又有概算的灵活性的混合方法.他们总结了以上的思想 产生出一个新数据库引擎,2005开了InfoBright公司.

InfiniDB

Column-Oriented DBMS:

如:

EmpId	Lastname	Firstname	Salary
1	Smith	Joe	40000
2	Jones	Mary	50000
3	Johnson	Cathy	44000

在row-oriented database中存储时被序列化成

1,Smith,Joe,40000;2,Jones,Mary,50000;3,Johnson,Cathy,44000;

在column-oriented DB中被序列化成:

1,2,3;Smith,Jones,Johnson;Joe,Mary,Cathy;40000,50000,44000;

区别在于:

1) 适合于运算于行多列少(一个列的较小集合)数据表查询.

2) 适合于更新整列的操作,可以不碰到其它列

3) 而row-oriented DB适合于更新有单行的多列.当一行很小时,只要一次磁盘操作就可取回整行

4) 而row-oriented DB适合于写入一个新行如果所有的列数据都提供的话

InfiniDB 是属于Column-oriented DB,由Calpont Corp.提供, GPL v2许可证. Open source

InfiniDB Features:

• Column-oriented, 适合查询
• 多线程设计.
• 自动的水平和横向分区.即行列都能分区
• 高并发:理论上无并发的限制,只受制于服务器的容量
• 高速数据装载器
• DML(可以视为是SQL的子集)支持 : 语句insert, update, delete
• Transactional支持:
• Crash恢复:
• 多版本并发控制(MVCC),保证所有的语句执行在一个特定的snapshot中,系统称之为(System Change Number, SCN). 这个称为Version Buffer的东西由内存(表明一个被修改的8块的hash表)和磁盘结构(默认4 个1G的文件, 散布在一台机器的多个驱动器上,(就象InfiniDB的数据文件一样)组成.
• 无需索引
• 低维护需求,诸如不需要实例化的视图和统计表来达到高性能
• Alter操作支持:add, drop
• 逻辑数据压缩，透明的，当前只通过去除重复数据的方法来压缩.以后会有针对列级别上物理数据的压缩.
• 性能自检：提供了SQL的分析和跟踪工具来查错和调优
• MySQL作为前端接口
• 运行在通用的硬件平台上
• 无许可证费用
• 商业智能(BI)工具兼容,因为用了MySQL的连接器，所有支持与MySQL兼容的BI工具

From InfiniDB Technical Overview White Paper:

对于有选择的查询，即只要求读取很少的列时，row-based DB要化5~10倍的物理IO相对于Column-based DB.

因此Row-based DB利用索引,水平分区,物理视图,摘要表和并行处理来提高性能.这同时也带来了负面问题.如空间,索引的选择,更复杂的更新增删操作.

Extract-Transform-Load (ETL)指数据处理的三个主要过程 : 从不同的来源抽取数据,变换及过滤数据(有时可能没有),装载数据到Data warehouse.

通常reporting DB与data warehouse是不同的,它们仅是一个Transactional系统的镜像,用于查询当前或历史数据.数据在这两个系统间装缷,复制,或Extract-Load

InfiniDB还有社区版和商业版之分,

商业版还有如下Feature:

• 大并行处理能力:能采用多台通用硬件平台的机器来达到性能线性增长.能动态加入节点
• 分布式的Share-nothing数据缓冲: 有一个逻辑上的cache分布在各参于的节点中.要求大内存
• 自动Failover
• 自动并发的Scale-out
• 自动软件打补丁管理

Architecture:

用户模块(User Module) ;　由小的MySQL实例和一些InfiniDB的掌控并发扩充的进程组成．本模块也负责将SQL查询分解在针对于＂性能模块＂的多个部分，最终本模块组装所有查询的结果给用户，可以有多个用户模块存在，这提供了高可用性配置及查询的负载平衡

性能模块(Performance Module) : 负责保存, 读取, 管理数据库,处理块请求并传回"用户模块", 本模块从磁盘或Cache(也就是前述的Share-nothing的数据缓冲)中读数据.分布式Cache就存在于本模块中,增加PM也就增加了Cache. (好象只有企业版(商业版?)有)

存贮(Storage)利用本地存贮或共享存贮(e.g. SAN (storage area network)), 可以是单台或多台服务器

单台机(Community Edition) Architecture

Enterprise (Commercial Edition) Architecture

用户查询处理流程:

1. 一个查询来到MySQL (InfiniDB Server), MySQL进行表操作从MySQL中得到一个查询执行计划
2. InfiniDB转换MySQL表对象为InfiniDB对象,并发给一个UM
3. UM转换MySQL执行计划并优化这些对象成为InfiniDB执行计划.UM决定这个计划的执行步骤和时机
4. UM咨询Extent Map,来得到适合查询的数据的位置
5. UM发送命令给PM,执行块IO操作
6. PM进行预过滤,连接处理,初始化聚合数据和发回结果给UM
7. UM进行最终结果聚合和合成最终结果
8. UM返回结果集给用户

InfiniDB的存贮思想:

Block : 8k的数据块,有Logical Block ID,大小不能定制,但预读的数目可以定制

Extent : 一个逻辑空间尺寸,存在于一个或多个的称为segment文件的物理文件中. extent大小受1)默认的行数2)一个列的数据类型, 如默认行数是8M,对一字节数据类型来说,Extent大小就是8M; 对于8字节数据类型来说,就是64M;对于可变长数据类型来说也是64M.当一个Extent满了,一个新的Extent就会被创建出来.

Segment File : 当一个Segment文件达到它的Extent包含的最大数目,一个新的Segment文件就会被创建出来

Partition : 与row-based DB不同之处在于它是一个逻辑上的对象.由一个或多个Segment文件组成. 一个列的partition数目是不限的.

HadoopDB

分析数据库(Analytical DB)占有39.8亿/146亿(27%)的数据库市场.递增率是10.3%

Share-nothing : 无共享,指数据分布在不同的主机器,数据横向分割,相互之间是独立的.

Analytical DB vendor leader : Teradata.

Share-nothing project :

Oracle, Exadata

Microsoft, Madison

以share-nothing布署的DBMS称为并行数据库(parallel database)

HadoopDB 由三部分组成

1) PostgreSQL (or MySQL) : database layer

2) Hadoop : communication layer

3) Hive : translation layer,给SMS planner用的轻量修改版.

相关的项目:

1) Yahoo, Pig

2) Microsoft, SOCPE

3) Open source, Hive

4) Greenplum, 永亚软件

5) Aster Data

4,5加入了写MapReduce方法的支持.

期望的特性:

Performance : 低消耗,低底线也是高性能的表现

Fault Tolerance : 对于Analytics DB来说,没有要开机重运行的查询. 分析的工作负担是只读查询而且无更新. 机器越多,配置越低,失败机率越高,

运行在主机类型差异的环境的能力 : 主机差异可能性很高,并发查询的速度取决于最差的主机的性能

灵活的查询接口: UDF (user defined functions), SQL, non-SQL 都需要

背景和当前解决方案的缺点:

Gamma ; Grace parallel DBMS : Fault Tolerance, 差异环境的运行能力不行.

MapReduce: Performace不行

HadoopDB:

组件:

1) Data Connector : 连接MySQL和PostgreSQL,并做为Hadoop中InputFormat的实现. 是DBMS与TaskTracker之间的接口

2) Catalog : DB的元信息,如连接参数,数据集的分布,副本的位置,分区等等. 当前是以XML文件形式保存在HDFS上,以后想以独立服务的形式提供.

3) DataLoader : 装载时通过一个指定的分区键重分区数据,将单点的数据分为小的多块分区或块(chunk),及以批量的装载多个单点的数据库以块的形式, 分为两个,Global Hasher和LocalHasher

GlobalHasher从存在HDFS上原文件读出数据,并分区成和Cluster结点一样多的块,(估计)还保存在HDFS上

LocalHasher从HDFS上读出一个分区,并切分成更小的块(有最大块大小的设置,推荐1G)保存在本地的数据库中.

两个Hasher用的相同的hash方法,它的不同之处在于确保每个块有相同的大小.

4) SQL到MapReduce,再到SQL (简称SMS planner)

Hive将一个SQL查询(HiveQL)变为多个operator,以有向无环图的形式组织,分别交由MapReduce的两个阶段执行.Hive中有简单的优化器来优化operator.

SMS planner采用相同的方式, 但在Map阶段,直接对本地数据库(本地的Chunk)进行SQL的查询,这是因为每个Slave(TaskTracker)都有安装的数据库,对Reducer来说还要分成多个operator,(这应该是采用了Hive的技术).

对于SMS planner来说,针对不同类型的分区有两种情况,一是分区时已保证相同的主键被分到了一个TaskTracker, 这时候只要一个Map的操作就能完成,二是分区时不能保证结果集的主键分到同一个TraskTracker, 这时候要Map和Reduce两个阶段一起工作.

HadoopDB安装布署:

1)Java 1.6

2)Hadoop 0.19.1

3)安装DBMS在每个slave结点上.

HBase 研究 -- HLog

HFile

==================================

HFile 有两个内部类

1) Reader

2) Writer

都是公开的静态的.

HFile结构:

Reader装入HFile通过 LoadFileInfo方法实现

1) 从文件尾读到Trailer

2) 在Trailer结构中读到FileInfo

3) 通过Trailer结构中读出data index

4) 通过Trailer结构中读出meta index

index 结构是由HFile.BlockIndex类实现

BlockIndex有三个数组,分别是

1) blockKeys byte[][]

2) blockOffsets long[]

3) blockDataSize int[]

1)是一个组合的Key

2,3) 指向了Value域. Value有可能是压缩的,因此,应从当前的blockOffset[]中读出到下个下一个blockOffset[]为止,读出的字节解压缩后大小应是blockDataSize[]中读出的值

HFile.Reader读出的data block可以cache在BlockCache中. BlockCache是一个接口,实现有LruBlockCache,SimpleBlockCache等,

对于BlockCache来说,存入一个block的方法是调用cacheBlock,参数是一个blockName字串,和一个用ByteBuffer类型表示的Buffer.

******************************************

HRegion

==================================

checkReadOnly()方法会抛异常.

2009/11/09

HLog 文件位于hdfs://主机:<9000>/hbase/.log/,<60020>,/

2009/11/11

TransactionalRegionServer

==================================

是HRegionServer的替代,实现了TransactionRegionInterface

与Transaction相关的方法是beginTransaction, delete, get, put, openScanner, commit, commitIfPossible, commitRequest,

与Transaction相关的有TransactionId (long),TransactionState.

TransactionId来自于客户端,HBaseBackedTransactionLogger.createNewTransactionLog()的返回值,从内部实现看是一个随机数, TransactionId是Client与Server之间最基本的联系

TransactionState在Server与Client端各有一个,Client是org.apache.hadoop.hbase.client.transactional.TransactionState. Client端的TransactionState是一个封装了TransactionId及相关的Region,也就是说一个Transaction关联多个Region.

Server端的TransactionState,是org.apache.hadoop.hbase.regionserver.transactional.TransactionState. 真正保存了Transaction的状态(PENDING(初始), COMMIT_PENDING(确定可以提交了), COMMITED(提交了) or ABORTED(放弃))),保存的所有读(Get的row key),所有的Scan(ScanRange), 所有的删除(Delete),以及所有的写(Put), 以及所有可能冲突的TransctionState的集合(set).

在Server端,Transaction是被委托到Region来进行的.Client与Server之间联系就是TransactionId,及一个确定的Region Name.即是通过TransactionalRegion类来实现.

TransactionRegion通过OCC机制来完成Transaction,但由于一个Transaction跨多个Region,所以,所有的Region必须都确保可以提交才能提交. 这是由Client端的org.apache.hadoop.hbase.client.transactional.TransactionManager来完成.如果在client端中间提交的过程中(也就是所有Region都确定可以提交一个Transaction后),有失败发生.则我们要重新决定Transaction,这个目前没有实现.

Transaction采用两阶段提交,1)prepareCommit, 2)doCommit.两阶段提交中,Get是即时的,也就是说,在prepareCommit之前就已经有结果返回了. 在Transaction中,对于一条特定的row, Get会读原表得到一个结果,并且当前Transaction有对同一row操作的Put(这个put 肯定1)在这个Get之前, 2)还未真正执行)的话,会读出这个Put并生一个合并的结果.

Put, Delete是在doCommit中执行的,并且没有考虑顺序问题(Bug).

TransactionalRegionServer在实例化时会创建一个Leases线程,用于做Transaction的监控.Leases线程会定时检查一个Lease是否过期,如果过期,则会回调这个Lease的Listener来发布这个消息. Leases采用了DelayQueue来操作Lease.

Transaction的Lease是在这个Region的Transaction创建时创建(TransactionRegion.beginTransaction), 对应的LeaseListener是TransactionLeaseListener, 如果Lease对应的Transaction是PENDING(初始)的,则直接设置它状态为ABORT,如果是COMMIT_PENDING, 则进行CommitPendingWaits计数,并更新Lease.如果CommitPendingWaits的计数到10(hard code的),则将这个Transaction通过Log做提交. 但由于全局Log没有实现,当前的做法是直接提交.

有关Scan --Scanner

==================================

HRegionServer实现的HRegionInterface中含有如下方法与Scanner相关

1) openScanner

2) 两个next函数来取回Scan结果

3) close来结束一个Scan

Scan的Handle是由openScanner返回的scannerId.

TransactionalRegionServer新加入了一个openScanner的方法,这个方法能接额外的transactionId参数.

Scan采用Scanner类来完成工作.即接口InternalScanner的实现类

InternalScanner的有两个方法:

1) next 用来取得下一个row的结果集一个KeyValue的list. (应是针对于多个Column)

2) close

InternalScanner用于返回一条row记录,包括了在scan中指定的多个column.

HRegionServer中用一个支持并发的HashMap来保存所有的InternalScanner,键即是scannerId(一个随机数)的转换成的字符串.

HRegionServer中采用的InternalScanner的实现类是RegionScanner.

RegionScanner内部委托KeyValueHeap来取得结果. RegionScanner对于传入的Scan,生成相应的多个KeyValueScanner.并与addictional的KeyValueScanner合并生成KeyValueHeap. 前者Scanner是从Store生成的,而一个Store取一个Column Family有关.另从Store取出的KeyValueScanner其实会从两个地方读取KeyValue的值 : 1)MemStore (Cache); 2) 多个StoreFile (真正的存贮文件,每个StoreFile对应于一个Column(Column Family : Qualifier)).

KeyValueScanner用于取得一个KeyValue.即一个Column的value,Column做为一个Key已经包括Row key, column family, qualifier, 及timestamp.

KeyValueHeap实现了KeyValueScanner及InternalScanner接口.内部有三个成员.一个当前的KeyValueScanner, 一个PriorityQueue存放其它的KeyValueScanner.以及一个比较器用来对PriorityQueue中的KeyValueScanner进行排序. 因此KeyValueHeap的工作原理是在内部的多KeyValueScanner中进行查找,以找出最前的一个Row相关的所有KeyValue对.来返回.

TransactionRegionServer中会另加入一个来自TransactionState中的Scanner,这个Scanner是PutScanner, PutScanner实现KeyValueScanner及InternalScanner接口.实例化时,PutScanner从当前的TransactionState中取得所有的Put并且放入KeyValueScanner的一个集合中.