基于Hadoop的企业数据仓库建设与创新刘汪根

基于Hadoop的企业数据仓库建设与创新2016/6/231www.transwarp.ioCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGG刘汪根大数据平台研发总监星环信息科技（上海）有限公司wayne.liu@transwarp.io企业级数据仓库架构OLAP

Analytics数据集市抽取Reporting转换清洗加载ETL数据仓库数据源数据服务CC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCIBTI

GC

IGTICTCGG2016/6/232www.transwarp.io一个典型的案例1. 单机数据库，单表上亿记录已经是存储、查询以及分析的最大上限2. 多维Cube数据膨胀块，单机无法存储，集中式存储昂贵3. 查询性能下降，进而导致稳定性下降4. 无法适应新业务的要求，如移动应用或者时效性要求SSIS+T-SQLSQLServer

DWSQL

ServerSSASOLAP

AnalyticsReportingSSRSCC

GGITITC问CG题IG：TCITGCITCGGITITCC

GGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGG2016/6/233www.transwarp.ioCC

GGITITCCG如IGTCI果TGCIT需CGGITI要TCC

GG融IITTCC合GGITIC大TGCIT数GCIGT据ICTCGG，IITTCC数GGITI仓TCCGI如GTCITG何CITCG构GITITCC建GGIIT呢TCCG？GITCITGCITGC

IGTICTCGG2016/6/234www.transwarp.io某互联网巨头的企业数仓架构SOURCEATAStructuraldataStreamingSourcesPortalTactical

ReportingBusiness

IntelligenceAnalyticsForecastingETL

ClustersMaster

ClusterTeradataStandby

ClusterTeradata<5

PBHBaseHiveKylinUn-structural

dataMySQLDaily

Sync-up MySQL报表区>50

PBTeradata

优点•完善的SQL支持，事务支持，提供OLAP分析功能•MPP执行引擎稳定，可以处理100TB以上的数据•管理工具比较完善缺点•单节点>百万$，并且在大数据规模可扩展性不足•无法存储非结构化数据，或者实时数据•无法有效支持数据挖掘类需求CC

GGIDTITCCGIGTCITGCITCGGITITCC

GGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGG2016/6/235www.transwarp.ioMPP+

Hadoop混合结构2016/6/236www.transwarp.io• 混合架构– 结构化数据入MPP，非结构化+实时数据数据入Hadoop– 核心BI报表由MPP生成，低优先级任务放在Hadoop上– OLAP服务由MPP提供，MPP上数据每天同步到Hadoop上– 为了保证SLA，需要有个严格的规则控制对MPP的资源使用• 优点– Hadoop作为补充，能够满足新型业务的需求，并且重用已经构建的传统MPP数仓系统• 缺点– 系统成本比较高，Teradata非常昂贵– 应用/开发人员需要详细的管理各个数据表的存储和计算方式，一个业务需要适配两套模型– 没有一个统一的引擎，应用使用跨系统的数据需要数据先同步– 实时业务开发门槛比较高，丢数据情况很常见，无HA– 开源Hadoop不支持事务，做全表级别的数据同步非常麻烦– Hadoop对SQL支持比较弱，通常要MapReduce、Spark以及脚本语言混合编程– Hadoop对数据的管理和开发支持很弱– 需要一个比较大的运维团队，提供架构支撑和数据开发CC

GGITITCCGIGTC–ITGC可IT以CG满GI足TI离TCC线批GG处IITT理CC与G在GI线TICOTLAGCPI分TG析C的IGT需ICT求C；GG两IIT套TCC系统GG互ITI相TC备CG份I，GTC提IT升GC数IT据CG的GI安TIT全CC性GGIITTCCGGITCITGCITGC

IGTICTCGGHadoop需要解决的关键问题稳定性、可靠性以及计算性能非常重要，当前Hadoop计算能力和稳定性都存在问题，开发和运维成本也很高SparkMllib技术门槛高，只能由数据科学家使用，普通的业务分析人员无法有效使用这些技术数仓能力与构建成本数仓批处理和OLAP数据挖掘能力数据一致性与数据同步支持实时、非结构化数据处理能力ACDE BETL等批处理业务是数仓的资源消耗大户，OLAP是保证BI性能的关键，Hadoop在这方面能力不足数仓平台需要能够处理实时数据和非结构化数据的能力，开发和运维实时类应用需要简单可靠跨系统间数据同步和一致性无法有效保证ETL的一致性对业务非常关键CGITCGITCGITC

GIT2016/6/237www.transwarp.ioCGITCGITC

GITCGITCGITC

GITCGITCGITC

GITCGITCGITCGITCGITC

GCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITC

GITCGCC

GGITITCC如GIGTCI何TGCIT解CGGITI决TCC

GGHIITTaCCdGGIoTICToGCpITG作CIGTIC为TCGG统IITTCC

一GGITITC数CGIGT仓CITGC的ITCGG关ITITCC键GGIITT问CCGGI题TCITGCITGC

IGTICTCGG2016/6/238www.transwarp.io星环科技团队的成果1.支持最全SQL，唯一支持PL/SQL3.支持分布式事务处理4.业界唯一分布式流式SQL 5.丰富的数据挖掘和机器学习算法6.运维无忧•

简单易用•

7*24小时不间断•降低流应用开发门槛，提高流应用开发效率•针对性的优化让StreamSQL比编程开发流应用性能更高•帮助用户零成本将传统业务逻辑变成流应用•

保证事务处理的ACIDBEGINTRANSACTION/COMMIT/ROLLBACK语法进行事务处理• 两阶段封锁协议可保证事务的完全可序列化•

多版本(快照)隔离可以保证只读事务的2.性能超群率先进入复杂数据分析的百TB时代2016/6/239www.transwarp.io•支持99%的SQL

2003语法Hadoop业界唯一支持OraclePL/SQLHadoop业界唯一支持DB2

SQL/PL•

帮助用户零成本迁移传统应用CC

G•GI新TITC应C用G的IGT开CI发TG成CIT本CG很低GITITCC

GGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGG高IT并ITC发C性GGIITTCCGGITCITGCITGC

IGTICTCGGInceptorPL/SQL

Compiler架构SQLParserSQLStatementsAbstractSyntax

TreeConstantConstantFoldingConFsotaldnint

gFoldingRDDDAGSQLNormalizerLogical

OptimizerCSEbytecodegenerationcolumnpruneroperatorprunerpartitionprunerpredicatepushdownPL/SQLASToptimizer

SQL2003JoinoptimizationsCBO

OptimizerPL/SQLControlfunctiondead

coderedundantinlining elimination eliminationCSEloopinvariantsCFG

Optimizer Parallel

OptimizercursorparallelizationTableStatisticsDAG

OptimizershufflereducerPhysicalPlanDAGSchedulersparktasksparktasksparktaskFirstPL/SQL

Compileron

Hadoop;98%Oracle

PL/SQLCompatibility.CGITCAnGaIlyTzCerGITCFlGowITGCrapGhITCGITCGITCGITCGITCGITCGI2016/6/2310www.transwarp.iohoistingTCGITCGITCGITCGITCGITCGITC

GCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITC

G050010002000250030001TB10TB30TB100TB

Transwarp

Inceptor’s

Performance

TPC-DS

Execution

Time

for

99

Queries

(in

minutes)可扩展的分布式计算技术Test

environment:•29worker

nodes•2CPUs,12Cores,E5-2620

v2•100GB

memory•Network:2X

1Gbps•Disks:12X

3TBORCFilesDataNodeORCFilesDataNodeORCFilesDataNodeORCFilesDataNodemetastore

1500InceptormasterExecutorExecutorExecutorExecutorTransactionManagerZK

Cluster星环SQLon

Hadoop已经能够高效处理100TB数据的复杂分析2016/6/2311www.transwarp.ioTranswarpInceptor’sPhysicalDeployment

Diagram单机 计算并行 计算分布 数据、计算均分布化MPI消息通讯模式

Map/Reduce计算模式C

GITCGITC

GITCGITCGITCGITCGITC

GITCGITCGITCGITCGITCGITCGITC

GITCC

GITCGITCGITCGITC

GITCGITC

GITCGITCGITCGITCGITC

GITCGITCGITC

GITCGG支持分布式事务处理采用多版本两阶段封锁协议实现可串行化快照隔离(SerializableSnapshot

Isolation)Transaction1

begin

transaction

selectmax(price)fromorderswhereage<

20

readvalueintolocalvariable

maxorder

updateorderssetprice=maxorder-1,

…

commitTransaction2

begin

transaction

…

updateorderssetprice=200whereid=

“007”

commit优点：1.

两阶段封锁协议可保证事务的完全可序列化；2.

多版本(快照)隔离可以保证只读事务的高并发性3.

分布式计算保证超高的吞吐率80000060000040000020000002016/6/2312www.transwarp.io1000000120000020000001800000160000014000001

column8

columns18

columnsCRUDPerformance(MERGE

INTO)Test

environment:•4worker

nodes•2CPUs,12Cores,E5-2620

v2•256GB

memory•Disks:6X

1TBCGITCGITC

GITCGITCGITCGITCGITCGITCGITC

GITCGITC

GITCGITCGITCGITCGITCGITC

GCGITCGITCGITCGITCGITC

GITCGITCGITC

GITCGITCGITCGITCGITC

G采用分布式内存进行交互式分析1WA2XB3YC4ZD5OE6PF7QG8RHHDFSStorage

LayerHDFSTextorORCorParquetFilesMemory

TierSSD

TierColumnarstoreSecondaryindexTable

formatSSDas

cacheHolodesk–AColumnarStoreIn-memoryoronSSDcache

layerFileSystem

APIInceptorServerExecutor

Executor

Executor

ExecutorColumnarStore

APICube

(D1,

D2,D3)Column

D1Column

D2Column

D3INDEXINDEXINDEX

INDEXColumn

M1Cube

(D1,

D2),(D2,D3),

(D1,

D3)ColumnarStore

APICube

(D1,

D2,D3)Column

D1Column

D2Column

D3INDEXINDEXINDEX

INDEXColumn

M1Cube

(D1,

D2),(D2,D3),

(D1,

D3)ColumnarStore

APICube

(D1,

D2,D3)Column

D1Column

D2Column

D3INDEXINDEXINDEX

INDEXColumn

M1Cube

(D1,

D2),(D2,D3),

(D1,

D3)ColumnarStore

APICube

(D1,

D2,D3)Column

D1Column

D2Column

D3INDEXINDEXINDEX

INDEXColumn

M1Cube

(D1,

D2),(D2,D3),

(D1,

D3)ZK

ClusterCube是OLAP分析的常用2016/6/2313www.transwarp.io技–术

SliceDiceRollupDrill

UpDrill

DownPivotCC

GGITITCCGIGTCOITfGf-CHITeCGapGITITCC

GGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGG支持SQL的流处理用StreamSQL对网站访问情况按URL统计CREATESTREAMaccesslog(ipSTRING,urlSTRING,time

TIMESTAMP)TBLPROPERTIES(“topic”=“accesslog”,“kafka.zookeeper”=“2:2181”);CREATETABLEresult(urlSTRING,count

INT);按消息时间以10s间隔统计网站访问情况INSERTINTOresultSELECTurl,count(1)FROMaccesslogGROUPBYtimeSTREAMWINDOWswas(SEPERATEDBYtimeLENGTH„10‟SLIDE„10‟

SECONDFORMAT„yyyy-MM-dd

hh-mm-ss.SSS‟);按系统时间以10s间隔统计网站访问情况INSERTINTOresultSELECTurl,count(1)FROMaccesslogGROUPBYtimeSTREAMWINDOWswas(LENGTH

„10‟SLIDE„10‟

SECOND);CGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITC

G2016/6/2314www.transwarp.ioCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITC

GStream架构图SourceManagerDistributedExecution

EngineApplication

ManagerSinkerSQL

CompilerStream

SQLODBCJDBCShellR

API ModuleAlgorithm

APIData

Mining2016/6/2315www.transwarp.ioStorage

ManagerComputing

LayerTranswarpHyperbase Transwarp

HolodeskPL/SQLandANSISQLonstreaming,canruncomplexOLAPanalysison

StreamKafka,socketandfilesourcesare

supportedProvideawriterpoolandimprovewritingthroughputintoHyperbase&

HolodeskParalleltasksschedulingforbetter

performanceOptimizedexecutionforwindowdata

analysisCGITCGITCGITC

GITCGITCGITCGITCGITCGITC

GITCMaGnaIgTeCrGITCGITCGITCGITC

GCGITCGITCGITCGGIITTCCGGITICTCGITCGITCGITCGITCGITCGITCGITCGITC

G用分析师熟悉的开发方式做客户行为分析训练数据采样民生银行2012年的04~09半年的交易流水，一共大约2亿条记录，506万消费频繁度消费水平美食爱好旅游爱好体育爱好电子爱好IT爱好年轻活力男性女性商人2016/6/2316www.transwarp.io开车一族电话达人差旅人士数GI据TC大小约80G。并行360度用户画像在2分钟内完成对506万独立持卡人的画像GITC

GITCCC

G个IT独C立GI持TC卡G人ITC，GITC

GGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGG数据挖掘产品DiscoverHDFSInceptorLocal

FileRDBMS模型获取统一接口层R存储算法接口分析层数据接口DataFrame(Table)

数据统一接口层机器学习库 深度学习库特征库 特征提取库行业方案Python2016/6/2317www.transwarp.ioCGITC

算法库CGITCGIGTCITGCITCGGITITCCGGIITTCCGGITICTGCITGCIGTICTCGGIITTCC

GGITITCCGIGTCITGCITCGGITITCCGGIITTCCGGITCITGCITGC

IGTICTCGGTDH数据仓库架构Holodesk数据集市ETL数据仓库SybaseJDBC/ODBC数据源数据服务OracleInceptorTranswarpManagerGuardianR

StudioData

DictionaryStreamDiscoverTranswarpDataAliveHyperbaseSQL

ServerPortalInformixDB2Tactical

Reportingsiness

IntelligenceAnalytics消息队列ForecastingCGITC

GITC2016/6/2318www.transwarp.ioGITCGITCGITC

GITCGITC

GITCGITCGITCGITCGITC

GITCGITC

GITBCu

GITCGITC

GCGITCGITCGITCGITCGITC

GITCGITCGITCGITCGITC

GITCGITCGITC

G某互联网公司ETL验证0.101.0010.000100200600500400300复杂ETL任务性能对比TDH Teradata Perf

Speedup/coreTDH2016/6/2319www.transwarp.io40worker

nodes•Network:

10Gbps•Disks:4X

1TBTeradata•6800H,3700

AMP•2CPUs,14Cores,

E5-2697•Network:

BYNETETL7tables,3factand4

dimension9.8TBin

total16complexSQL,join/groupby/mergeinto/insert

coveredTDH完全胜任Teradata运行的复杂ETL任务，并且单CPU的效能是Teradata的两倍。• 使用TDH独立构建的数据仓库完全可以取代混合部署C

GITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITCGITC•2GCPITUsC,

10GCoIrTesC,

E5G-2I6T20Cv2GITC

GCGITCGITCGITCGITCGITCGITCGITCGITC

GITCGITC•1G28G