大数据开发实战：Stream SQL实时开发一

　　一、流计算SQL原理和架构

　　　　流计算SQL一般是一个类SQL的声明式语言，主要用于对流式数据(Streams)的持续性查询，目的是在常见流计算平台和框架（如Storm、Spark Streaming、Flink、Beam等）的底层API上，

　　经过使用简易通用的的SQL语言构建SQL抽象层，下降实时开发的门槛。

　　　　流计算SQL的原理其实很简单，就是在SQL和底层的流计算引擎之间架起一座桥梁---流计算SQL被用户提交，被SQL引擎层翻译为底层的API并在底层的流计算引擎上执行。好比对Storm

　　来讲，会自动翻译成Storm的任务拓扑并在Storm集群上运行。

　　　　流计算SQL引擎是流计算SQL的核心，主要负责对用户SQL输入进行语法分析、语义分析、逻辑计划生成、逻辑计划执行、物理执行计划生成等操做。而真正执行计算的是底层的流计算平台。

　　　　不一样于离线任务，实时的数据是不断流入的，因此为了使用SQL来对流处理进行抽象，流计算SQL也引入了“表”的概念，不过这里的表是动态表。

　　　　流计算SQL的架构以下：

　　　　

　　　　SQL层：流计算SQL给用户的接口，它提供过滤、转换、关联、聚合、窗口、select、union、split等各类功能。

　　　　SQL引擎层：负责SQL解析/校验、逻辑计划生成优化和物理计划执行等。

　　　　流计算引擎层：具体执行SQL引擎层生成的执行计划。

　　二、流计算SQL：将来主要的实时开发技术

　　　　目前流计算SQL在各个计算框架的进度和支持力度不一。

　　　　Storm SQL还只是一个实验性的功能。Flink SQL是Flink大力推广的核心API。Flink是一个原生的开源流计算引擎，并且目前尚未其它开源流计算引擎能提供比Flink 更优秀的流

　　　　计算SQL框架和语法等，因此Flink SQL实际上在定义流计算SQL的标注。

　　　　阿里云Stream SQL 的底层就是Flink引擎（实际是Blink,也就是Alibaba Flink）,能够认为Blink是Flink的企业版本，

　　三、Stream SQL

　　　　阿里云提供了Stream SQL 开发的完整环境，包括Stream SQL语法、IDE开发工具、调试及运维等。下面具体介绍概念和语法

　　　　3.一、Stream SQL 源表

　　　　　　Stream SQL 一般将源头数据抽象为源表，就像一个Storm任务必须至少定义一个spout,一个Stream SQL 任务必须至少定义一个源表。

　　　　　　定义Stream SQL 源表的语法以下：

　　　　　　CREATE TABLE tablename

　　　　　　(columnName dataType [,columnName dataType]*)

　　　　　　[WITH (propertyName=propertyValue [,propertyName=propertyValue ] * )];

　　　　　　以下面的例子建立了一个datahub类型的源表

　　　　　　create table datahub_stream(

　　　　　　　　name varchar,

　　　　　　　　age BIGINT,

　　　　　　　　birthday BIGINT)

　　　　　　　　with (

　　　　　　　　　　type ='datahub',

　　　　　　　　　　endPoint =‘http://dh-et2.aliyun-inc.com’,

　　　　　　　　　　project='blink-datahub_test',

　　　　　　　　　　topic ='test_topic_1',

　　　　　　　　　　accessId =0i70RRFJD1OBAWAs',

　　　　　　　　　　accessKey ='yF60EwURseo1UAn4NinvQPJ2zhCfHU',

　　　　　　　　　　startTime='2018-08-20 00:00:00'

　　　　　　　　　　);

　　　　　　其中的type表示流式数据的源头类型，能够为datahub,也能够为日志或消息中间件等，type下面的各个参数类型的不一样而不一样，它们共同肯定了此type的某个源头类型。

　　　　　　此外，阿里云Stream SQL底层流计算引擎是Flink/Blink,所以其支持水位线机制。

　　　　　　定义水位线的语法以下：

　　　　　　WATERMARK  [watermarkName] FOR <rowtime_field>

　　　　　　AS withOffset(<rowtime_field>,offset)

　　　　　　好比WATERMARK FOR rowtime AS withOffset(rowtime,4000)就对源头数据列rowtime定义了固定延迟4s的水位线。

 

　　　　3.2 、Stream SQL 结果表

　　　　　　有源表，就是结果表，Stream SQL定义结果表的语法以下：

　　　　　　CREATE TABLE tablename

　　　　　　(columnName dataType [,columnName dataType]*)

　　　　　　[WITH (propertyName=propertyValue [,propertyName=propertyValue ] * )];

　　　　　　Stream SQL的结果表支持各类类型，包括相似MySQL的RDS、相似HBase的TableStore、相似消息队列的MessageQueue的，下面以RDS来介绍Stream SQL 结果表的具体语法：

　　　　　　create table rds_output(

　　　　　　　　id int,

　　　　　　　　len int,

　　　　　　　　content varchar,

　　　　　　　　primary key(id,len)

　　　　　　) with (

　　　　　　　　　type ='rds',

　　　　　　　　　url='jdbc:mysql:XXXXXX',

　　　　　　　　　tableName='test4',

　　　　　　　　　userName='test',

　　　　　　　　　password='xxxx'

　　　　　　);

　　　　　　在上述代码中，结果表的type不一样，相应后面的其它参数也不同，具体可用参考阿里云帮助文档。

　　　　　3.三、Stream SQL维度表

　　　　　　　　流计算SQL的维度表数据一类特殊的外部数据，相对流数据来讲，他比较稳定且变化缓慢，是静态或准静态数据，做为join / left outer join的右表使用。须要特别注意的是，

　　　　　　维度表在流计算中不容许做为from 后面的数据存储。流计算中对于from子句后对接的数据存储必定是流式数据存储，即 select * from dim_table是不被容许的。

　　　　　　　　阿里云Stream SQL中没有专门为维度表设计的DDL语法，使用标准的create table语法便可，可是须要额外增长一行PERIOD FOR SYSTEM_TIME的声明，这行声明定义了

　　　　　　维度表的变化周期，即代表该表是一张会变化的表。

　　　　　　　　一个简单的维度表定义实例以下，type后面的语法相似源表定义，

　　　　　　　　CREATE TABLE white_list (

　　　　　　　　　　id varchar,

　　　　　　　　　　name varchar,

　　　　　　　　　　age int,

　　　　　　　　　　PRIMARY key(id),　　--用做维度表，必须有声明的主键

　　　　　　　　　　PERIOD FOR SYSTEM_TIME ---定义了维度表的变化周期

　　　　　　　　　　) with (

　　　　　　　　　　　　type = 'xxx',

　　　　　　　　　　　　。。。

　　　　　　　　　　）；

　　　　　　

　　　　 　3.四、Stream SQL 临时表

　　　　　　在实际的实时开发中，常常发现业务逻辑的复杂性使得只用一个Stream SQL来完成全部的业务逻辑基本是不可能的，而必须拆分为多个SQL共同完成，此时就须要定义中间临时表（

　　　　　　在阿里云Stream SQL 中也叫view，即视图）。在Stream SQL中定义临时表的语法以下：

　　　　　　CREATE VIEW viewName

　　　　　　[ (columnName[,columnName]*])]

　　　　　　AS queryStatement;

　　　　　　但须要注意的是，Stream SQL临时表仅用于辅助计算逻辑表达的内存逻辑中间状态，其物理是并不存在，也不会产生数据的物理存储。固然，临时表也不占用系统空间。一个临时表的例子

　　　　　　以下：

　　　　　　　　CREATE VIEW largeOrders(r, t, c, u) AS

　　　　　　　　SELECT rowtime, productId, c, units

　　　　　　　　FROM Orders;

　　　　　　

　　　　3.五、Stream SQL DML

 　　　　　　Stream SQL语法和SQL标准语法绝大部分都是相同的，下面仅着重介绍insert操做

　　　　　　insert操做的语法：

　　　　　　INSERT INTO tableName

　　　　　　[ ( columnName[,columnName]* )]

　　　　　　queryStatement;

　　　　　　流计算不支持单独SELECT操做，当前在执行SELECT查询以前必须执行INSERT操做将结果保存起来。同时，须要注意的是，一个SQL文件支持多个源表输入和多个结果表输出。

　　　　　　只有result表和tmp表能够执行INSERT操做，且每张表只能执行一次INSERT操做，dim 表和stream表不能执行insert操做。

　　　　　　普通的select操做是从几张表中读数据，但查询的对象也能够是另外一个select操做，也就是子查询，但要注意子查询必须加别名，实例以下：

　　　　　　insert into result_table

　　　　　　select * from （

　　　　　　　　　　　　select t.a,   sum(t.b) AS sum_b,   from t1 t   

　　　　　　　　　　　　group by t.a

　　　　　　　　　　　　) t1

　　　　　　where t1.sum_b>100;

 

　　　参考资料：《离线和实时大数据开发实战》