第13讲实时流处理引擎slipstream

实时流处理引擎Slipstream范颖捷|2018年10

1234StreamSQLDDLStreamSQLDML1234551StreamSQLvs.批处批量处调度延处理延流处流式处低延

Slipstream基 批处理计算框SparkFlink流式计算框基 驱－Storm－Flink基于微型批处理（Micro-－Spark－Storm

Slipstream基 流式计算框架存在的

Slipstream基

问题丢数无结无响

定融 驱动与微批处理的实时流计算引分布式流式SQL引特微批模式 驱动模式 －在同一套系统里，可以根据业务需求，通过参数配置和SQL改写，灵活切换两种流处理模支持分布式流式－支持通过SQL创建并操作流，并支持在流上

Slipstream基

Slipstream基特强大的优化器提－在一些条件下，采用StreamSQL可以获得比编程方式更高的性能，因为Slipstream做了特殊优化，编极高的易用－只要有编写普通SQL的经验，就可以写出高效、稳定、安全的流处产品化程度－StreamSQL作为一个通用的接口显著提升了产品化程迁移成本－用户原有的业务逻辑很多是通过SQL实现的，系统迁移时只需修改SQL，迁移成本大

Slipstream基Slipstream的三 概Stream（数据流StreamJob（流任务－对一个或多个Stream进行计算，并将结果写入一张表Application（流应用－一组业务逻辑相关的StreamJob的集stream

stream

Slipstream基Stream（数据流InputStream（输入流 驱动（Event-driven）模式：以单条数据被InputStream接收 Stream（数据流DerivedStream（衍生流①含义：Stream变形是从已有Batch计算得到新Batch的过

Slipstream基

Slipstream基Stream（数据流DerivedStream（衍生流＊WindowStream：通过窗口变形得到的Derived

Slipstream基Stream（数据流DerivedStream（衍生流 ①含义：每得到一条数据就对其进行变形，得到Derived③窗口变形（多数据变形）：对一个时间窗口（Window）内的多条数据进行计算得到新数据的过

Slipstream基StreamJob（流任务要让StreamSQL执行计划，需要有相应的Action操作来触发收数据，接收来的数据经过一系列DerivedStream的变形最 入一张表，供用户查 Application（流应用合理使用Application划分StreamJob可Stream的StreamJob 若使用同一个InputStream，则每个

Slipstream基 StreamSQLvs.DML语句的运行机制普通SQL－提交SQL后，用户需等待SQL执行结束，期间命令被持续阻塞，无法执行 StreamSQL：背景运 －执行StreamSQL的DML语句会立即返回结查询结果的输出普通SQL：查询结果或者显示在Console，或者通过StreamSQL：用户必须显式地指定查询结果输出到某个地 持续运行的SQL无法直接跟Console交－查询结果通常会插入到表中，如：InsertIntoresult_tableSelect

Slipstream基 2Stream管理操ShowStreams：列出所有的DescribeStream：查看Stream信

StreamSQL

StreamSQL创建InputStream：完整语CREATESTREAM<stream_name>(<col_name><data_type>,<col_name><data_type>,...)[ROWFORMATDELIMITEDFIELDSTERMINATEDBY'<delimiter1>‘COLLECTIONITEMSTERMINATEDBYMAPKEYSTERMINATEDBY'<delimiter3>']TBLPROPERTIES(["topic"="<topic_name>",]②["source"="kafka",]③["kafka.zookeeper"="<ip:port>",]④["kafka.broker.list"="<ip:port>",]⑤["partitions"="<int>,<int>,...",]⑥["tocol"="<protocol>",]⑦["transwarp.consumer.sasl.jaas.config"="jass_string",]⑦["batchduration"="<int>",]⑧["<key>"="<value>",]⑨["ducer.<property_key>"="kafka_producer_property_value",]⑪

StreamSQL创建InputStream：完整语④<ip:port>指定TDH集群中的一个Zookeeper节点的ip和端 ⑧设置Batch的长度，整数，单位为毫秒。该属性只能在

StreamSQL创建InputStream：完整语 创建InputStream：数据源为发布给KafkaTopic的消息通过InputStream进入一个InputStream只能接收一个Topic中的消息，多个InputStream可以指向同一个

StreamSQL

StreamSQL创建InputStream：数据源为创建从指定Topic中接收数据的CREATECREATESTREAMdemo(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','"topic"="kafka.zookeeper"="kafka.broker.list"=创建从指定Partition中接收数据的CREATESTREAMCREATESTREAMdemo(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','"topic"="partitions"="kafka.zookeeper"="kafka.broker.list"=

StreamSQLCREATECREATESTREAMsocket_stream(idINT,name"source"="host"="port"=创建InputStream：数据源为创建InputStream：复杂类型/*Kafka/*Kafka数据格式 2,9|nineCREATESTREAMstruct_stream(idINT,s1STRUCT<a:INT,b:STRING>)ROWFORMATDELIMITEDFIELDSTERMINATEDBY',‘COLLECTIONITEMSTERMINATEDBY

StreamSQL创建InputStream：复杂类型/*Kafka/*Kafka数据格式 2,2|4|6|8CREATESTREAMarray_stream(idINT,a1ARRAY<STRING>)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','COLLECTIONITEMSTERMINATEDBY'|';/*Kafka/*Kafka数据格式：1,job:800|team:60 CREATESTREAMmap_stream(idINT,m1MAP<STRING,INT>)ROWFORMATDELIMITEDFIELDSTERMINATEDBYCOLLECTIONITEMSTERMINATEDBY'|'MAPKEYSTERMINATEDBY创建InputStream：复杂类型/*Timestamp默认格式：yyyy-MM-ddKafka/*Timestamp默认格式：yyyy-MM-ddKafka数据格式：1,2015-01-15 2,2077-11-1523:16:10CREATESTREAMstudent_stream(idINT,t1

StreamSQL创建DerivedStream：数据源为使用已有的Stream创建DerivedCREATECREATESTREAM<stream_name>ASSELECTCREATESTREAMfemale_name_streamASSELECTnameFROMemployee_stream CREATESTREAMemployee_window_streamASSELECT*FROMemployee_streamSTREAMWINDOWw1AS(length'12'secondslide'4'second);列出所有Stream与查看Stream信SHOWSHOWDESCRIBE|DESC[EXTENDED|FORMATTED]STREAM修改InputStream：重命 ALTER ALTERSTREAM<old_name>RENAMETO

StreamSQL修改InputStream：修改属ALTERSTREAM<stream_name>SETTBLPROPERTIES("key"="value");修改InputStream：增加ALTERALTERSTREAM<stream_name>ADDCOLUMNS(nametype[,name1type1修改InputStream：替换ALTERALTERSTREAM<stream_name> CECOLUMNS(nametype[,name1type1删除InputDROPDROPSTREAM StreamJob是触发StreamSQL执行的Action，一般具有插入结果表语StreamJob主 StreamJobLevel的配置参数，以及对应的StreamJob管理操CreateStreamJob：创建ShowStreamJobs：列出所有持久化的DescribeStreamJob：查看StreamJob信AlterStreamJob：修改DropStreamJob：删除

StreamSQL

StreamSQL创建CREATESTREAMJOB<streamjob_name>AS("<sql_action>")查看StreamJob信列出所有持久化的SHOWSHOW修改ALTERALTERSTREAMJOB<streamjob_name>SET删除DROPSTREAMJOBDROPSTREAMJOB

StreamSQLStreamJob运行时管StreamJob作为StreamSQL运行时的基本单元，也是实 的基本单/*/*启动当前Application下的所有StreamJob*/STARTSTREAMJOBS;/*启动当前Application下的某个StreamJob*/STARTSTREAMJOB<s1>;/*查看当前Application下的所有StreamJob*/LISTSTREAMJOBS;LISTSTREAMJOB/*停止当前Application下的所有StreamJob*/STOPSTREAMJOBS;/*停止当前Application下的某个StreamJob*/STOPSTREAMJOB<s1>;

StreamSQL 在测试时，用户登录 （如Beeline），在自己建的Application下编写业务逻当业务需要上线时，可以用一个ProdId，具有严格的权限控制，并对应一个Application Application管理操CreateApplication：创建ShowApplications：列出所有DescribeApplication：查看Application信AlterApplication：修改DropApplication：删除

StreamSQL创建CREATECREATEAPPLICATIONWITHDESCRIBE|DESCDESCRIBE|DESCAPPLICATION列出所有SHOWSHOW修改ALTERALTERAPPLICATION<app_name>SET删除DROPAPPLICATION DROPAPPLICATION

StreamSQLApplicaiton用户运行任何StreamJob之前需要进入对应的默认情况下，当前的Application是/*/*进入Application*/USEAPPLICATIONapp1;/*显示正在运行的Application*/LISTAPPLICATIONS; 3

StreamSQL流应用通常会对一个窗口（时间间隔）内的数据做多表关联、聚合或统非窗口计默认以系统时间为基准，每隔一段时间（微批）或每产生一条数据 ）触发一次计窗口计算：时间作为窗口切分单时间切分：在SQL中指定数据流中的某个时间字段，以该时间字段为基准切分窗系统时间切分：在SQL中不指定时间字段，以系统时间为基准切分窗窗口类滑动窗－示例：对于窗口长度为2s，滑动间隔为1s的滑动窗口，[0s,2s)为第一个窗口，[1s,3s)为第二个窗口[2s,4s)为第三个窗口，依次类

StreamSQL窗口类滑动窗 ，跳动窗口是滑动窗口Length=Slide的特－示例：以2s为间隔的跳动窗口，[0s,2s)为第一个区间，[2s,4s)为第二个区间，依次类窗口切分方系统时间切分：以流处理引擎的系统时间为基准切分窗CREATECREATESTREAMs1(idINT,nameSTRING,tsINSERTINTOt1SELECT*FROMs1STREAMWINDOWw1AS(LENGTH'2'SECONDSLIDE'1' 窗口切分方时间切分：以数据流中的指定时间字段为基准切分窗

StreamSQLSETstreamsql.use.eventtime=true;CREATESTREAMs1(idINT,nameSTRING,tsINSERTINTOt1SELECT*FROMs1STREAMWINDOWw1AS(LENGTH'2'SECONDSLIDE'1'时间切分：自定义时间格式（默认时间格式为yyyy-MM-dd指定格式为“yyyy-MM-ddHH:mm”，不包含秒级信息*/CREATESTREAMs1(idINT,nameSTRING,tsINSERTINTOt1SELECT*FROMs1STREAMWINDOWw1AS(LENGTH'2'MINUTESLIDE'1' 窗口切分方时间切分：支持String类

StreamSQLCREATECREATESTREAMs1(idINT,nameSTRING,ts时间切分与系统时间切分的区

StreamSQLCREATESTREAMASSELECTnameFROMemployee_streamCREATESTREAMASSELECT*FROMSTREAMWINDOWw1AS(LENGTH'12'SECONDSLIDE'4'输出到StreamSQL支持将查询结果输出到另一个方式一：将一个流原样输出到另一个CREATECREATESTREAMs1(idINT,nameSTRING);CREATESTREAMs2(idINT,nameSTRING);INSERTINTOs2SELECT*FROMs1;方式二：将一个流变换为另一个

StreamSQL输出到/*创建输入流CREATESTREAMemployee_stream(idINT,name /*创建HyperbaseCREATETABLEhyper(row_keydata_type,col_name1data_type,col_name2,data_type,STOREDBYWITHSERDEPROPERTIES("hbase.columns.map column_family:column_qualifer_1,column_family:column_qualifier_2,...")/*将流转换后插入HyperbaseINSERTINTOhyperSELECT*FROMemployee_stream

StreamSQL输出到/*创建两个KafkaTopic*/CREATESTREAMs1(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBYCREATESTREAMs2(idINT,letterSTRING)ROWFORMATDELIMITEDFIELDSTERMINATEDBY/*将s1输出到s2INSERTINTOs2SELECT*FROM 4

驱动的流处含以单条数据被InputStream接收 ，逐 数据，并立刻加工处理，最后输来一条处理一特相比微批模式（TDH4.8之前 驱动模式的延迟更低，在延迟敏感的场景中表现更开 驱动模第一步：让Slipstream引擎运行 驱动模式 驱动模式（TDH5.1之后的默认值），mapred代表微批（TDH5.1之前的默认值第二步：让StreamJob运行 驱动模式=

驱动的流处/*/*开启Job驱动模式SET/*创建Kafka输入流CREATESTREAMs1(scoreINT,nameSTRING)/*创建InceptorCREATETABLEt1(scoreINT,nameINSERTINTOt1SELECT*FROMs1;将流表中的数据写入Inceptor普通 4.2

驱动的流处在WindowStream上做聚合后写入Inceptor普通/*开启Job 驱动模式SET/*开启窗口的 SETstreamsql.use.eventtime=true;/*创建WindowStreamCREATESTREAMs1(nameSTRING,scoreINT,tsSTRING)CREATESTREAMws1ASSELECCT*FROMs1STREAMWINDOW(LENGTH‘4’SECONDSLIDE‘2’/*先对WindowStream做聚合