2024基于storm的实时计算架构

实时流计算应用开发框架-天罡提纲1.背景2.业界3.产品介绍4.架构设计5

.

s

q

u

i

r

r

e

l Q

L6.实践经验1.1

业务背景Big

data数据量膨胀业务快速变化，商业模式的创新SNS,移动互联网用户体验个性化，实时化…1.2

离线计算vs.流计算From

<IBM

InfoSphere

Streams:

Harnessing

Data

in

Motion>1.3

数据分析演进趋势2.1业界流计算引擎2.2

storm介绍最新版本storm

0.74主要特性：-适用场景广泛-可伸缩性高-保证无数据丢失-异常健壮-容错性好一些问题：-编程门槛对普通用户较高-框架无持久化存储-框架不提供消息接入模块-storm

ui功能简单-跨topology的bolt复用-nimbus单点-topology不支持动态部署2.3

storm架构3.1

阿里巴巴业务需求特征目前需求归类统计业务关键指标，客观反映当前的业绩现状，比如网站活动监控需求特征1跟踪业务指标的变化趋势，出现异常波动，能智能报警2分钟级延迟；丌能漏算；丌能错算；统计时长为当天；分钟级延迟；丌能漏算；丌能错算；统计时长为当天；业务闭环运营中的实时数据应用，比如事件营销，触发式服务3秒级延迟；允许漏算；丌能错算；计算过程复杂（规则多）；实时推荐4秒级延迟；允许漏算；丌能错算；不推荐系统交互频繁；实时数据信息服务5秒级延迟；丌允许漏算；丌能错算；计算

过程复杂（指标定义复杂，指标个数多）；部分指标统计时长跨天；3.2天罡-产品定位天罡：实时流计算应用开发框架深刻理解实时业务需求，提供实时计算的完整应用开发框架屏蔽系统复杂性，可配置方式即可完成消息源接入类SQL工具，封装引擎系统，降低实时计算任务的开发难度完整的任务管理系统，提供任务配置、发布、管控一条龙服务强大的运维管理系统，监控系统、任务、数据的状态，适时报警3.3天罡Features简单统计计算(包括时间窗口)多流join计算(动态)容错，事务中间状态持久化统一消息接入支持类SQL支持数据类型：int/long/string/double…支持schema支持join,distinct,group

by,count,top

N支持常规函数-to_char

,substr…3.4天罡-核心概念数据采集数据接入流式计算数据输出处理流程Job信息processingExtract

jobMessage

sourceMessage

topicMessage

topicMessageMessageDataMessageProcessing

job由messageextractservice提供功能凼数由类SQL组件实现业务逻辑天罡系统里的相关概念说明：一个message

source（相当亍database）包括1个以上的message

topic(相当亍table)，一个message

topic由1条以上的message（相当亍record）组成。天罡系统有二类Job，分别称为extract

job和processingjob。一个extract

job负责一个message

topic的消息接入，一个

processingjob由1个以上的message

topic参不计算。一个message

topic可以被多个processingjob引用。Processingjob的输入内容称为message，计算的结果输出，称为data。4.1天罡-技术方案Engine

state

storageerosa精卫实时logStream

processing

engine(storm)HBASE…Squrriel

QLtable

join

distinct

arg消息处理流向SQL解析组件SQL执行计划组件元数据管理算子组件supervisorboltnimbuszookeperspoutLocal

buffer

cachePersistence

storageDataoutputserviceHBASEDBmessageconnectormessageadaptorMessageextractservicemetaMSCmanagerMessagestoragecenterfile…4.2天罡-功能模块划分Message

extract

serviceMessage

store

centerStream

processing

engineEngine

state

storageData

output

serviceData

subscriber

serviceGlobal

Control

SystemMonitor

&

AlarmJob

Control

CenterJob

Configure

AssistantCommand

Line

ClientLogMetadata

Centersquirrel

QL4.3天罡-功能模块关系应用开发系统模块运维管控报警MessageextractserviceMessagestore

centerLogEnginestate

storageProcessing

operatorinterfaceStream

processingengineJob

controlcenterJob

Configure

AssistantDataoutput

serviceMonitor（system，service，job，data）AlarmMetadata4.4

Message

extract

serviceExtract

jobmessage连接器，能够接入db、log、file、mqmessage适配器，提供字段选择、格式转换功能实现Job的配置和管理接口服务分布式调度控制管理负载管理Job启动，停止规则配置– Keyword

searchMessage

Extract

ServiceZookeeperRainbowMajordomoMajordomo……1023103179123……Job4.5

Message

store

center统一消息存储方式基于Metamorphosis：淘宝一款类似Kafka强大的通用消息中间件Pull模式的MQ高吞吐量Meta

spout支持消息顺序Message

read

interfaceMessage

write

interfaceMSC

manager4.6

Engine

state

storage中间状态持久化需求场景join聚合原始消息存储容错，事务消重聚吅JoinDistinctBolt存储策略层LocalBufferHBaseRedis……4.7

Global

Control

System系统、任务等配置信息的统一存储管理中心提供用户体验良好的任务研发和配置服务任务管理中心，直观反应当前任务执行状态，能够中止、暂时、重设任务监控：系统（内存，cpu，网络，

io等），功能（redis，meta，storm等），任务，数据质量storm集群管理5.1

stormtopology开发效率过滤中间状态存储容错输出…聚合简单统计场景（中文站UV,PV统计）：select

a,b,count(c),count(distinct

d)

from

x

where

a=1

group

byto_char(a,'yyyymmdd'),bTopologyBuilder

builder

=

newTopologyBuilder();builder.setSpout(1,

cbuuvpvSpout(),

10);builder.setBolt(2,

cbupvuvBolt(),3).shuffleGrouping(1);……5.2

squirrel

QL架构CLIwebsqlSQL解析SQL执行计划metadata数据源selectselectjoin数据源数据源count(*)输出IDEtablejoindistinctarg……输出算子5.3

squirrel

QL

解析流程SQLDsComponentJoinComponmentstormDsstormJoinstormOptspoutboltAGGDistProjSeleOptComponment5.4

squirrel

QL聚合SELECT

uid,COUNT(order) FROM

tableGROUP

by

uidtableCOUNTAGGDs:Projection（uid，order）Opt:Goup

by

uidSchema(uid,count)5.5

squirrel

QL

JOINSELECTc.country,COUNT(o.order)

FROMcustormercorder

o

onjoinc.uid=

o.uidGROUP

BY

c.countryCUSTOMERORDERCUSTOMER_ORDERCOUNTAGGDs:Projection:uid,countryDs:Projection:uid

,orderJoinOpt:Schema(country,order)CountaggOperation:Grouby

countrySchema(country,count)5.6

squrriel

QL&Hive

QLHive

QLSqurriel

QL数据存储HDFSHBASE数据格式用户自定义用户自定义数据更新不支持不支持索引有（0.8版之后增加）无执行MapReduceTopology可扩展性高（UDF，UDAF,UDTF）高6.1

实践经验中间状态读写策略缓存,超时6秒LRU命中率6.2实践经验(Cont.)消息顺序如何保证？全序，偏序设置保序key消息中间件meta存储顺序保证在同一个task（fields

grouping,自定义hash

grouping）Bolt

ASpoutTaskBolt

BMetaQ消息分区StreamGrouping相同key的消息hash分区6.3

实践经验(Cont.)时效与准确性间平衡纯流计算(时序不严格，注重时效性，准确性可 通过批量计算补偿)精确计算(牺牲一定时效，接近准确)消息保序transactional

topolgoy

(transaction

id)消