2. 实时计算

Last updated 4 years ago

Was this helpful?

2. 实时计算

本用户手册供在大数据套件（以下简称TBDS）中使用实时计算（以下简称EasyCount）的用户使用，主要包括三个部分，第一部分主要介绍EasyCount中间件的一些基本概念；第二部分介绍任务管理操作；第三部分介绍案例。

本文档适用于具有一定SQL基础的计算机专业人员，概况一下读者对象适用于以下几类读者：

流计算开发人员
数据仓库工程师
数据分析人员
后台技术运营人员

EasyCount系统，用一句话介绍就是使用SQL描述业务实时计算的需求，并将SQL转化为基于storm的topology实时计算平台。

相对于传统的SQL，EasyCount加入了窗口的概念，使得数据可以一直保存在内存中，由此可以快速进行大量内存计算，EasyCount的输出结果为数据流在某一时刻的计算结果。EasyCount是建立在Storm基础上的类SQL查询语言，它解决了Storm原生API使用复杂，上手难度高，很多基本功能缺失的问题，提升了流处理产品的易用性。

EasyCount设计目标就是，用纯粹的SQL语句再加上一些命令，就可以完成所有的任务发布以及执行，这样，就可以通过EasyCount Web门户配置和管理实时计算任务。对于有一定SQL基础的用户，只需要掌握一些EasyCount SQL比较特殊的语法，比如窗口或者临时表定义的语法，就可以配置出可运行的实时计算任务，大大降低了实时计算上手难度。

2.1 EasyCount简介

2.1.1 EasyCount系统介绍

EasyCount从kafka或hippo读取数据同时关联第三方维表数据（mysql、hbase）进行实时分析，计算结果数据有多重分发途径，可以回流到TDBank，也可以发送到KV，DB（TPG，MYSQL），ES，以及HBASE等。以下为easycount运行示意

EasyCount是基于Storm-yarn平台基础上的通用统计模块，通过简单的配置即可生成一个可以运行在Storm-yarn上的Topology，进行实时统计计算。Storm-yarn解决了大集群的资源分配和资源隔离问题，为EasyCount能够运行在更大的集群上去铺平了道路，为EasyCount的应用开发提供了无限的想象空间。

EasyCount实时计算平台主要分为两大块组成：第一块是EasyCount web 门户，用户管理easycount脚本和easycount任务，第二块是EasyCount core,SQL Driver负责将SQL的解析成AST、DataSource负责计算数据源接入的实现、DataSink负责将计算结果会写到配置的结果表、Operator负责计算算子的实现、UDF负责系统自定义函数的实现、config负责系统的配置管理、monitor负责监控指标的统计及上报。

2.1.2 EasyCount基本概念

表：EasyCount系统处理的基本单元是“表”，也就是说需要有明确的schema定义的结构化数据，不过相对于传统数据库的表中的字段定义，EasyCount中的表支持map，list，struct以及binary等复杂数据类型。根据数据表的功能，EasyCount系统中的表分为四种，流水表、维表、结果表、临时表。
- 流水表：是一组（无穷）元素的集合，流水表上的每个元素都属于同一个schema；每个元素都和逻辑时间有关；即流包含了元组和时间的双重属性。流水表与传统数据表不同，在于传统数据表的一次计算是针对整张表数据进行的，而流水表数据的计算只针对当前时间的数据进行，最多是针对截止到当前的所有数据进行的计算。流水表数据的统计是实时进行的，根据我们大多数应用需求场景，流水表一般都是按照某个小的时间粒度进行数据统计的，例如10s，1min，5min等。基于此，EasyCount系统在进行实时统计计算的时候，要求数据中必须有一个时间字段作为协调，如果数据中确实没有时间字段，那么就EasyCount系统按照接收到数据的时间进行协调。流水表是EasyCount实时计算系统的重要数据源,目前系统支持的流水表有kafka和hippo。
- 维表：维表是指那些静态数据表，我们在和用户进行需求沟通的时候，发现大量的配置需求，也就是说流水表中的信息是不完全的，在实际的统计中需要关联一些配置信息，之后再进行数据统计。这些配置信息，有时候需要修改，但是通常变化不大。为此EasyCount系统将这类数据表抽象为维表，维表数据一般存储在KV系统中，如果维表数据量较小的话，也可以存储到DB甚至以配置文件的方式抽象为内存表，直接从内存中关联。维表目前支持以下几种：HBASE、MYSQL。
- 结果表：结果表是用来存储计算结果的表。结果表可以是kafka、hippo这样的流水表，也可以是hbase这样的KV表，还可以是mysql这样的db。具体的计算结果流向哪里由业务根据计算结果的数据量及使用情况做权衡。
- 临时表：当多组计算逻辑使用同一结构的数据集时可以将该数据集使用With关键字定义为临时表。
Window：窗口（window）是流处理中解决事件的无边界及流动性的一种重要手段，把事件流在某一时刻变成静态的视图，以便进行类似数据库表的各种查询操作。Easycount系统中的窗口需要和聚合函数配合使用。传统sql对于聚合函数的定义都是针对整张数据表进行的，然而对于实时计算来说，这是不可能的，数据是流式进入的，聚合统计只能做到对当前某个时间段的数据进行。因此这里定义了聚合窗口（AGGR INTERVAL）的概念，是指进行数据聚合所采用的时间粒度。假设聚合窗口为60s，那就表示每一分钟进行一次聚合计算，聚合计算的结果是针对这1分钟数据进行的。这里提到时间窗口，那么这个时间是按照什么标准进行协调的呢，默认情况下，如果不使用coordinate by子句的话，系统按照接收到数据时的系统时间进行协调，如果用户指定了coordinate by子句的话，就按照用户指定的字段进行协调。注意，用户指定的字段必须是一个毫秒数，如果不是的话可以通过函数转化为毫秒数。
在easycount sql窗口有三种类型，聚合窗口、滑动窗口和累加窗口。累加窗口和滑动窗口，这两个窗口和聚合窗口一样是两个聚合时间粒度。不过这里要求累加窗口和滑动窗口必须是聚合窗口的整数倍。
- 普通聚合：和传统聚合函数一致，对每个聚合窗口进行一次聚合计算。
上述的累加聚合和滑动窗口聚合的具体实现是比较耗费资源的，因此一般建议累加窗口和滑动窗口不要比聚合窗口大太多，但是在某些场景下，又无法实现，因此提供了一个新的全局累加功能，ACCUGLOBAL，这个功能就是一个累加聚合功能，不过区别是，启动这个功能以后，所有的聚合函数都是被设置为累加功能，并且同一条子查询中的滑动窗口聚合的功能被关闭。
表达式：符号和运算符的一种组合，easycount解析引擎处理该组合以获取单个值。简单表达式可以是常量、变量或者函数，可以用运算符将两个或者多个简单表达式联合起来构成更复杂的表达式。

2.1.3 EasyCount S-QL介绍

S-QL：Stream Query Language，流式查询语言，这里面的S和标准SQL中的S含义不同，为了便于区分，使用S-QL表达EasyCount系统中的流式查询语言。下表直接给出S-QL的语法规范。

语法约定

语法说明

示例

备注

回车

代表一条sql逻辑结束

With (Select ...),(Select ...)

定义临时表，多个临时表直接使用逗号分隔。

WITH (SELECT ... FROM...)tmpa, (SELECT ..... FROM .... ) tmpb

INSERT INTO

将计算结果写入到结果表

INSERT INTO xxx SELECT .... FROM tmpa, INSERT INTO yyy SELECT ..... FROM tmpb

COORDINATE BY column_x

指定按照某个时间字段划分统计窗口，默认值为系统时间。

COORDINATE BY unix_timestamp(dtevetTime, 'yyyy-MM-dd HH：mm：ss')*1000

按照dtevetTime这个时间字段划分统计窗口

AGGR

定义普通聚合窗口

WITH AGGR INERVAL 60 SECONDS聚合每隔60秒做一次统计输出

累加窗口和和滑动窗口必须以普通聚合窗口为基础，时间跨度为普通聚合窗口的整数倍。

ACCU

定义累加窗口

WITH ACCU INERVAL 120 SECONDS累加窗口每隔120秒做一次统计输出

同上

定义滑动窗口

WITH SW INERVAL 120 SECONDS 滑动窗口120秒做一次统计输出

同上

示例(以下为一条带有滑动、累加、聚合窗口的sql示例):

INSERT INTO test_result SELECT activityId, count(1) pv, countd_hllp(uiUin ACCU) huv, countd_hllp(uiUin SW) fmuv, from_unixtime(AGGRTIME DIV 1000, "yyyy-MM-dd HH：mm：ss") FROM test_src_kafka GROUP BY activityId COORDINATE BY unix_timestamp(dtevetTime, 'yyyy-MM-dd HH：mm：ss')*1000 WITH AGGR INTERVAL 60 SECONDS WITH ACCU INTERVAL 3600 SECONDS WITH SW INTERVAL 300 SECONDS

通常编写一条S-QL，可以按照如下步骤进行：

将所有的中间查询结果，通过WITH语法，以子查询的方式作为临时表，写在最前面，并用逗号分隔；
使用INSERT语法将计算结果写入目标表，多条INSERT语句使用逗号分隔。

可见，一条S-QL语句支持多个输入，多个输出，用户可以在一条S-QL内部充分发挥。S-QL能有这样的支持功能，主要原因是EasyCount系统中提交一条S-QL以后，任务将一直运行，其所占用的系统资源一直不会被释放，这在系统资源总数固定的条件下，系统能够承载的总任务数目是有限的，因此提交一个S-QL需要相当的谨慎，一条S-QL应该可以做更多的事情，从总体上减少系统资源的占用。多条S-QL示例：

with (select iActivityId, hllp(uiUin) uvb, from_unixtime(AGGRTIME DIV 1000, "yyyy-MM-dd HH:mm:ss") ts,
concat_ws('-', 'd', from_unixtime((AGGRTIME DIV 1000), 'yyyy-MM-dd 00:00:00'), cast(iActivityId as string)) dk
from src GROUP BY iActivityId
COORDINATE BY unix_timestamp(dteventTime, 'yyyy-MM-dd HH:mm:ss')*1000 WITH AGGR INTERVAL 60 SECONDS) tmp, 
(select iActivityId, hllp_merge(tmp.uvb, dim.uvball) uvball, tmp.dk k, ts from
tmp left join dim on tmp.dk=dim.k) jd
insert into dim with k as KEY select jd.k k, jd.uvball uvball from jd,
insert into dest select jd.iActivityId, hllp_get(jd.uvball) duv, jd.ts from jd

2.1.3 EasyCount数据类型

1. 基本数据类型

数据类型

Description

Example

TINYINT

1个字节（8位）有符号整数( 从-128 到 127), 后缀 Y 用来表示小范围的数字

10Y

SMALLINT

2字节（16位）有符号整数(从-32,768 到 32,767) , 后缀S用来表示一个egular descriptive number

10S

INT

4字节（32位）有符号整数(从-2,147,483,648 到 2,147,483,647)

BIGINT

8字节（64位）有符号整数(从-9,223,372,036,854,775,808 到 9,223,372,036,854,775,807),后缀为 L

100L

FLOAT

4字节（32位）单精度浮点数,范围在1.40129846432481707e-45 to 3.40282346638528860e+38 (正负值)，暂时还不支持科学计数法。用它进行存储会非常接近数字值

1.2345679

DOUBLE

8字节（64位）双精度浮点数，范围在（4.94065645841246544e-324d 到 1.79769313486231570e+308d]正负值]）。暂时还不支持科学计数法，用它来存储会非常接近数字值[numeric values]

1.2345678901234567

DECIMAL十进制

它的范围在1039 - 1 to 1 - 1038之间，DECIMAL数据类型存储数据的精确值,它的默认定义格式是decimal(10,0).decimal(a,b)中a代表小数点左边的最大位数，b代表小数点右边的最大位数

DECIMAL (3,2) for 3.14

BINARY

它只支持与STRING类型的转换，反之亦然。

1011

BOOLEAN

TRUE or FALSE

TRUE

STRING

它使用单引号（'）或者双引号（"）来表达包含的字符串。Hive 使用 C 语言格式的字符串，最大溢出大小在 2G左右。

'Books' or "Books"

DATE

用来指定年,月,日。格式是 YYYY-MM-DD。它的范围从0000-01-01 to 9999-12-31

'2013-01-01'

TIMESTAMP

从Hive 0.8.0开始便支持该类型，它用来描述指定的年,月,日,时,分,秒,毫秒。格式是YYYY-MM-DD HH：MM：SS[.fff...]

'2013-01-01 12：00：01.345'

2. 复杂数据类型

EasyCount有3个主要的复杂数据类型： ARRAY, MAP, 和 STRUCT。这些数据类型是建立在基本数据类型基础之上。STRUCT是一个 Record 类型，允许包含任意类型的字段。复杂数据类型允许嵌套类型：

复杂数据类型

描述

语法示例

ARRAY

数组是一组具有相同类型和名称的有序变量的集合。这些变量成为数组的元素，每个数组元素有一个编号，而且从0开始。例如：fruit[0]='apple'

['apple','orange','mango']

MAP

一组无序的键/值对。键的类型必须是原子的，值可以是任何类型，同一个映射的键的类型必须相同，值得类型也必须相同。 Map是一组无序的键值对元组的集合，使用数组表示法可以访问元素。例如，如果某个列的数据类型是 Map，其中 Key->value paris 对应的是'first'->'John' 和 'last'->'Doe'. 那么可以通过字段名 ['last'] 获取最后一个元素： fruit['last']='Doe'.

map('first', 'John', 'last', 'Doe')

STRUCT

一组命名的字段。字段类型可以不同。和 C 语言中的 struct 或者"对象“类似，都可以通过"点"分隔符访问元素内容。例如{val1, val2, val3, ....},默认情况下, STRUCT 字段名可以是col1,col2,...你可以通过structs_name.column_name 来访问具体的值： fruit.col1=1.

info structinfo.name获取nameinfo.age获取age

2.1.4 EasyCount表达式

表达式是符号和运算符的一种组合，easycount解析引擎处理该组合以获取单个值。简单表达式可以是常量、变量或者函数，可以用运算符将两个或者多个简单表达式联合起来构成更复杂的表达式。

1. 通用表达式

通用表达式可以出现在select子句中，也可以出现在where或者group by子句中。表达式的优先级从高到底如下表所示:

表达式

说明

+, -

正数，负数之类一元操作符

*, /

乘、除

+, -

二元操作符

=, <>,,=， !=

比较操作符

IS [NOT] NULL, [NOT] LIKE, [NOT] BETWEEN,[NOT] IN

是否为空之类判断表达式

AND、OR

逻辑表达式：多个条件之间是"且"或者"或"的关系

2. 特殊表达式

FOREACH & EXECUTE处理复杂数据类型时，可以使用foreach和execute语法进行循环和迭代处理。

2.1.5 EasyCount函数

EasyCount内部提供了很多函数给开发者使用，包括数学函数，类型转换函数，条件函数，字符函数，聚合函数，表生成函数等等,其中大部分函数继承hive,另外也有一部分EasyCount系统的自定义函数。

注明：__innertable(1000) 每个1000ms输出一次，用于测试使用，innertable中没有真实数据，可以用于验证函数。

1. 数值函数

取整函数： round，语法： round(double a)，返回值： BIGINT
说明：返回double类型的整数值部分（遵循四舍五入）
举例：select round(3.1415926) from __innertable(1000) ;
结果：3
指定精度取整函数： round，语法： round(double a, int d)，返回值： DOUBLE
说明：返回指定精度d的double类型
举例：select round(3.1415926,4) from __innertable(1000) ;
结果：3.1416
向下取整函数： floor，语法： floor(double a)，返回值： BIGINT
说明：返回等于或者小于该double变量的最大的整数
举例：select floor(3.1415926) from __innertable(1000);
结果：3
向上取整函数： ceil,语法： ceil(double a),返回值： BIGINT
说明：返回等于或者大于该double变量的最小的整数
举例：select ceil(3.1415926) from __innertable(1000);
结果：4
向上取整函数：ceiling，语法： ceiling(double a)，返回值： BIGINT
说明：与ceil功能相同
举例：select ceiling(3.1415926) from __innertable(1000);
结果：4
取随机数函数：rand，语法： rand(),rand(int seed)，返回值： double
说明：返回一个0到1范围内的随机数。如果指定种子seed，则会得到一个稳定的随机数序列
举例：select rand() from __innertable(1000);
结果：0.5577432776034763
举例：select rand(100) from __innertable(1000);
结果：0.7220096548596434
举例：select rand(100) from __innertable(1000);
结果：0.7220096548596434
自然指数函数：exp，语法： exp(double a)，返回值： double 说明：返回自然对数e的a次方
举例：select exp(2) from __innertable(1000);
结果：7.38905609893065
自然对数函数：ln，语法： ln(double a)，返回值： double
说明：返回a的自然对数
举例：select ln(7.389) from __innertable(1000);
结果：2.0
以10为底对数函数：log10，语法： log10(double a)，返回值： double
说明：返回以10为底的a的对数
举例：select log10(100) from __innertable(1000);
结果：2.0
以2为底对数函数：log2，语法： log2(double a)，返回值： double
说明：返回以2为底的a的对数
举例：select log2(8) from __innertable(1000);
结果：3.0
对数函数：log，语法： log(double base, double a)，返回值： double
说明：返回以base为底的a的对数
举例：select log(4,256) from __innertable(1000);
结果：4.0
幂运算函数： pow，语法： pow(double a, double p)，返回值： double
说明：返回a的p次幂
举例：select pow(2,4) from __innertable(1000);
结果：16.0
幂运算函数：power，语法： power(double a, double p)，返回值： double
说明：返回a的p次幂,与pow功能相同
举例：select power(2,4) from __innertable(1000);
结果：16.0
开平方函数：sqrt，语法： sqrt(double a)，返回值： double
说明：返回a的平方根
举例：select sqrt(16) from __innertable(1000);
结果：4.0
二进制函数：bin，语法： bin(BIGINT a)，返回值： string
说明：返回a的二进制代码表示
举例：select bin(7) from __innertable(1000);
结果：111
十六进制函数：hex，语法： hex(BIGINT a)，返回值： string
说明：如果变量是int类型，那么返回a的十六进制表示；如果变量是string类型，则返回该字符串的十六进制表示
举例：select hex(17) from __innertable(1000);
结果：11
举例：select hex(‘abc’) from __innertable(1000);
结果：616263
进制转换函数：conv，语法： conv(BIGINT num, int from_base, int to_base)，返回值： string
说明：将数值num从from_base进制转化到to_base进制
举例：select conv(17,10,16) from __innertable(1000);
结果：11
绝对值函数：abs，语法： abs(double a) abs(int a)，返回值： double int
说明：返回数值a的绝对值
举例：select abs(-3.9) from __innertable(1000);
结果：3.9
举例：select abs(10) from __innertable(1000);
结果：10
正取余函数：pmod，语法：pmod(int a, int b),pmod(double a, double b)，返回值： int double
说明：返回正的a除以b的余数
举例：select pmod(9,4) from __innertable(1000);
结果：1
举例：select pmod(-9,4) from __innertable(1000);
结果：3
正弦函数： sin，语法：sin(double a)，返回值： double
说明：返回a的正弦值
举例：select sin(0.8) from __innertable(1000);
结果：0.7173560908995228
反正弦函数： asin，语法：asin(double a)，返回值： double
说明：返回a的反正弦值
举例：select asin(0.7173560908995228) from __innertable(1000);
结果：0.8
余弦函数： cos，语法： cos(double a)，返回值： double
说明：返回a的余弦值
举例：select cos(0.9) from __innertable(1000);
结果：0.6216099682706644
反余弦函数： acos语法：acos(double a)，返回值： double
说明：返回a的反余弦值
举例：select acos(0.6216099682706644) from __innertable(1000);
结果：0.9
positive函数： positive语法：positive(int a), positive(double a)，返回值： int double
说明：返回a
举例：select positive(-10) from __innertable(1000);
结果：-10
举例：select positive(12) from __innertable(1000);
结果：12
negative函数： negative语法：negative(int a), negative(double a)，返回值： int double
说明：返回-a
举例：select negative(-5) from __innertable(1000);
结果：5
举例：select negative(8) from __innertable(1000);
结果：-8
2. 日期函数
UNIX时间戳转日期函数：from_unixtime，语法： from_unixtime(bigint unixtime[, string format])，返回值： string
说明：转化UNIX时间戳（从1970-01-01 00：00：00 UTC到指定时间的秒数）到当前时区的时间格式
举例：select from_unixtime(1493864893,'yyyy-MM-dd HH：mm：ss' ) from__innertable(1000);
结果：2017-05-04 10：28：13
获取当前UNIX时间戳函数：unix_timestamp,语法： unix_timestamp(),返回值： bigint
说明：获得当前时区的UNIX时间戳
举例：select unix_timestamp() from __innertable(1000);
结果：1493864893
日期转UNIX时间戳函数：unix_timestamp，语法： unix_timestamp(string date)，返回值： bigint
说明：转换格式为"yyyy-MM-dd HH：mm：ss"的日期到UNIX时间戳。如果转化失败，则返回0。
举例：select unix_timestamp('2017-05-04 10：28：13') from __innertable(1000);
结果：1493864893
指定格式日期转UNIX时间戳函数：unix_timestamp，语法： unix_timestamp(string date, string pattern)，返回值： bigint
说明：转换pattern格式的日期到UNIX时间戳。如果转化失败，则返回0。
举例：select unix_timestamp('20170504 10：28：13','yyyyMMddHH：mm：ss') from __innertable(1000);
结果：1493864893
日期时间转日期函数：to_date，语法： to_date(string timestamp)，返回值： string
说明：返回日期时间字段中的日期部分。
举例：select to_date('2017-05-04 10：03：01') from __innertable(1000);
结果：2017-05-04
日期转年函数： year，语法： year(string date)，返回值： int
说明：返回日期中的年。
举例：select year('2013-12-08 10：03：01') from __innertable(1000);
结果：2013 select year('2012-12-08') from __innertable(1000);
结果：2012
日期转月函数： month，语法： month (string date)，返回值： int
说明：返回日期中的月份。
举例：select month('2011-12-08 10：03：01') from __innertable(1000);
结果：12
日期转天函数： day，语法： day (string date)，返回值： int
说明：返回日期中的天。
举例：select day('2017-05-04 10：03：01') from __innertable(1000);
结果：4
日期转小时函数： hour，语法： hour (string date)，返回值： int
说明：返回日期中的小时。
举例：select hour('2017-05-04 10：03：01') from __innertable(1000);
结果：10
日期转分钟函数： minute，语法： minute (string date)，返回值： int
说明：返回日期中的分钟。
举例：select minute('2017-05-04 10：03：01') from __innertable(1000);
结果：3
日期转秒函数： second，语法： second (string date)，返回值： int
说明：返回日期中的秒。
举例：select second('2017-05-04 10：03：01') from __innertable(1000);
结果：1
日期转周函数：weekofyear，语法： weekofyear (string date)，返回值： int
说明：返回日期在当前的周数。
举例：select weekofyear('2011-12-08 10：03：01') from __innertable(1000);
结果：49
日期比较函数： datediff，语法： datediff(string enddate, string startdate)，返回值： int
说明：返回结束日期减去开始日期的天数。
举例：select datediff('2012-12-08','2012-05-09') from __innertable(1000);
结果：213
日期增加函数： date_add，语法： date_add(string startdate, int days)，返回值： string
说明：返回开始日期startdate增加days天后的日期。
举例：select date_add('2012-12-08',10) from __innertable(1000);
结果：2012-12-18
日期减少函数： date_sub，语法： date_sub (string startdate, int days)，返回值： string
说明：返回开始日期startdate减少days天后的日期。
举例：select date_sub('2012-12-08',10) from __innertable(1000);
结果：2012-11-28

3. 字符串函数

字符串长度函数：length，语法: length(string A)，返回值: int
说明：返回字符串A的长度
举例：select length('abcedfg') from __innertable(1000);
结果：7
字符串反转函数：reverse，语法: reverse(string A)，返回值: string
说明：返回字符串A的反转结果
举例：select reverse(abcedfg’) from __innertable(1000);
结果：gfdecba
字符串连接函数：concat，语法: concat(string A, string B…)，返回值: string
说明：返回输入字符串连接后的结果，支持任意个输入字符串
举例：select concat(‘abc’,'def’,'gh’) from __innertable(1000);
结果：abcdefgh
带分隔符字符串连接函数：concat_ws，语法: concat_ws(string SEP, string A, string B…)，返回值: string
说明：返回输入字符串连接后的结果，SEP表示各个字符串间的分隔符
举例：select concat_ws('-','abc','def','gh') from __innertable(1000);
结果：abc-def-gh
字符串截取函数：substr,substring，语法: substr(string A, int start),substring(string A, int start)，返回值: string
说明：返回字符串A从start位置到结尾的字符串
举例：select substr('abcde',3) from __innertable(1000);
结果：cde
字符串截取函数：substr,substring，语法: substr(string A, int start, int len),substring(string A, intstart, int len)，返回值: string
说明：返回字符串A从start位置开始，长度为len的字符串
举例：select substr('abcde',3,2) from __innertable(1000);
结果：cd
字符串转大写函数：upper,ucase，语法: upper(string A) ucase(string A)，返回值: string
说明：返回字符串A的大写格式
举例：select upper('abSEd') from __innertable(1000);
结果： ABSED
字符串转小写函数：lower,lcase，语法: lower(string A) lcase(string A)，返回值: string
说明：返回字符串A的小写格式
举例：select lower('abSEd') from __innertable(1000);
结果：absed
去空格函数：trim，语法: trim(string A)，返回值: string
说明：去除字符串两边的空格
举例：select trim(' abc ') from __innertable(1000);
结果：abc
正则表达式替换函数：regexp_replace，语法: regexp_replace(string A, string B, string C)，返回值: string
说明：将字符串A中的符合java正则表达式B的部分替换为C。注意，在有些情况下要使用转义字符,类似oracle中的regexp_replace函数。
举例： select regexp_replace('foobar', 'oo|ar', '-') from __innertable(1000);
结果：f-b-
正则表达式解析函数：regexp_extract,语法: regexp_extract(string subject, string pattern, int index),返回值: string
说明：将字符串subject按照pattern正则表达式的规则拆分，返回index指定的字符。
举例：select regexp_extract('foothebar', 'foo(.*?)(bar)', 1) from__innertable(1000);
结果：the
URL解析函数：parse_url,语法: parse_url(string urlString, string partToExtract [, stringkeyToExtract]),返回值: string
说明：返回URL中指定的部分。partToExtract的有效值为：HOST, PATH, QUERY, REF, PROTOCOL, AUTHORITY, FILE, and USERINFO.
结果：facebook.com
结果：v1
json解析函数：get_json_object,语法: get_json_object(string json_string, string path),返回值: string
说明：解析json的字符串json_string,返回path指定的内容。如果输入的json字符串无效，那么返回NULL。
举例：select get_json_object('{"fruit":apple,”ower”:”tim”}’,'$.owner')
结果：tim
空格字符串函数：space语法: space(int n)返回值: string
说明：返回长度为n的空格字符串
重复字符串函数：repeat，语法: repeat(string str, int n)，返回值: string
说明：返回重复n次后的str字符串
举例：select repeat('abc',5) from __innertable(1000);
结果：abcabcabcabcabc
首字符ascii函数：ascii，语法: ascii(string str)，返回值: int
说明：返回字符串str第一个字符的ascii码
举例：select ascii('abcde') from __innertable(1000);
结果：97
分割字符串函数: split，语法: split(string str, stringpat)，返回值: array
说明:按照pat字符串分割str，会返回分割后的字符串数组
举例： select split('abtcdtef','t') from __innertable(1000);
结果：["ab","cd","ef"]
集合查找函数:find_in_set，语法: find_in_set(string str, string strList)，返回值: int
说明:返回str在strlist第一次出现的位置，strlist是用逗号分割的字符串。如果没有找该str字符，则返回0
举例：select find_in_set('ab','ef,ab,de') from __innertable(1000);
结果：2

4. 条件函数

If函数: if，语法: if(boolean testCondition, T valueTrue, T valueFalseOrNull)，返回值: T
说明: 当条件testCondition为TRUE时，返回valueTrue；否则返回valueFalseOrNull
举例：select if(1=2,100,200) from __innertable(1000);
结果：200
举例：select if(1=1,100,200) from __innertable(1000);
结果：100
nvl函数: nvl，语法: nvl(T value, T default_value)，返回值: T
说明：如果value值为NULL就返回default_value,否则返回value
举例：select nvl(null,100) from __innertable(1000);
结果：100
isnull函数: isnull，语法: isnull(T value)，返回值: true,false
说明：如果value值为NULL就返回true,否则返回false
举例：select isnull(null) from __innertable(1000);
结果：true
isnull函数: isnotnull，语法: isnotnull(T value)，返回值: true,false
说明：如果value值为NULL就返回true,否则返回false
举例：select isnotnull(null) from __innertable(1000);
结果：false
条件判断函数：CASe,语法: CASE WHEN a THEN _a [WHEN b THEN _b]* [ELSE _c] END,返回值: T
说明：如果a为TRUE,则返回_a；如果b为TRUE，则返回_b；否则返回_c
举例： select case when 1=2 then 'tom' when 2=2 then 'mary' else'tim' end from __innertable(1000);
结果：mary

5. 类型转换函数

类型转换函数: cast，语法: cast(expr as )，返回值: Expected "=" to follow "type"
说明:返回array类型的长度
举例：select cast(1 as bigint) from _innertable(1000);
结果：1

6. 集合函数

Map类型长度函数: size(Map)，语法: size(Map)，返回值: int
说明:返回map类型的长度
举例：select size(map('100','tom','101','mary')) from _innertable(1000);
结果：2
Array类型长度函数: size(Array)，语法: size(Array)，返回值: int
说明:返回数组类型的长度
举例：select size(array(‘aa’,’bb’)) from _innertable(1000);
结果：2
7. 聚合函数
个数统计函数: count,语法: count(col),返回值: int
说明: count(expr)统计检索出的行的个数，返回指定字段的个数
总和统计函数: sum语法: sum(col),返回值: double
说明: sum(col)统计结果集中col的相加的结果；sum(DISTINCT col)统计结果中col不同值相加的结果
举例： select sum(t) from lxw_dual;
结果：100
平均值统计函数: avg，语法: avg(col)，返回值: double
说明: avg(col)统计结果集中col的平均值；avg(DISTINCT col)统计结果中col不同值相加的平均值
举例：select avg(t) from lxw_dual;
结果：50
最小值统计函数: min，语法: min(col)，返回值: double
说明:统计结果集中col字段的最小值
举例： select min(t) from lxw_dual;
结果：20
最大值统计函数: max，语法: maxcol)，返回值: double
说明:统计结果集中col字段的最大值
举例： select max(t) from lxw_dual;
结果：120
8. 扩展函数
个数统计函数: countd,countd_hllp，语法: countd(col),countd_hllp(col),返回值: int
说明: count(col)去重统计检索出的行的个数，返回指定字段的个数。该函数为非精确去重统计，精确度在99.5%左右。
去重合并统计函数: hllp_merge，语法: hllp_merge(a,b),参数为binary,返回值: binary
说明:去重合并统计函数是将两个二进制集合去重合并生成以一个新的集合。
获取去重后的结果函数：hllp_get,语法 hllp_get(a),范围在为bigint
说明：获取去重后的统计结果

以上三个函数都是基于HyperLogLog算法的实现，为了做大规模去重统计，降低存储空间，精确度上有一点误差。

SQL示例：

with (select iActivityId, hllp(uiUin) uvb, from_unixtime(AGGRTIME DIV 1000, "yyyy-MM-dd     HH:mm:ss") ts, 
concat_ws('-', 'd', from_unixtime((AGGRTIME DIV 1000), 'yyyy-MM-dd 00:00:00'), cast(iActivityId as     string)) dk 
from src GROUP BY iActivityId 
COORDINATE BY unix_timestamp(dteventTime, 'yyyy-MM-dd HH:mm:ss')*1000 WITH AGGR     INTERVAL 60 SECONDS) tmp, 
(select iActivityId, hllp_merge(tmp.uvb, dim.uvball) uvball, tmp.dk k, ts from 
tmp left join dim on tmp.dk=dim.k) jd 
insert into dim with k as KEY select jd.k k, jd.uvball uvball from jd,
insert into dest select jd.iActivityId, hllp_get(jd.uvball) duv, jd.ts from jd
[tabledesc-1]
table.name=src
table.fields=iActivityId,int,:dteventTime,string,:uiUin,string,
table.field.splitter=|
[tabledesc-dimtable-tde]
table.name=dim
table.fields=k,string,:uvball,binary,
table.field.key=k
[tabledesc-destination]
table.type=tpg
table.name=dest
table.fields=iActivityId,int,:duv,bigint,:dteventTime,string

2.2 操作指引

EasyCount-web门户主要分为两块，一个是任务管理模块，用于对easycount 任务进行管理，包括easycount 任务的创建、删除、更新、启动、停止、任务导入导出；另外一个是库表管理模块，主要是编辑保存经常使用的库表以便于后面创建任务的时候直接选择，减少任务库表信息的重复编辑。

登录套件的门户网站，点击实时计算模块，进入easycount模块。

2.2.1 如何新建一个实时计算任务？

点击新建任务。点击“任务管理”进入任务列表页面，点击新建任务按钮，进入新建任务页面；
填写任务信息。新建任务的信息分为三块：任务信息、库表信息和脚本信息；
- 任务信息：任务信息中填写任务的基本配置信息，包括任务名称、任务描述、选择任务的责任人及所属项目。扩展配置numWorkers代表任务在storm上需要启动多少个worker。
- 库表信息：库表信息是配置脚本信息中需要使用的表信息，库表分为源表、目标表和维表三类。
  源表：只能为流水表，代表计算数据的主要来源。
  维表：是一些静态数据，在计算过程需要关联的表。
  库表信息可以直接新增，也可以从已有的库表模板中导入（库表模板介绍详见下文），每个库表信息填写完整后可以点击测试验证该库表是否连通，通过方可使用。 注意项:
  - 在申请使用Hippo表时需要配置生产组和消费组权限，在使用kafka表时需要配置生产组权限。Hippo权限申请参考《数据接入使用手册》；
  - Kafka topic和HBase表的权限分配需要联系管理员申请，权限申请步骤为：进入运维中心模块，进入访问管理页面，申请所需权限；
  - 以上所需的表的权限申请通过之后才能在SQL中使用，否则在任务运行过程中会有权限异常抛出；
  - 脚本可通过预编译预先查看脚本是否能编辑通过。
运行任务。脚本编辑通过后点击保存，进入任务列表页面启动任务。点击运行任务后，任务会异步下发到storm集群，任务的状态会每隔30s刷新一次。
任务导出。为了将已有的任务方便的迁移到其他平台，可以选择任务点击任务导出，任务脚本会打包下载，解压后可以看到任务脚本。
任务导入：将已有的任务脚本导入到系统生成任务。

2.2.2 如何创建一个库表模板？

库表模板是用户自主编辑保存经常使用的库表以便于后面创建任务的时候直接导入，减少任务库表信息的重复编辑，节约时间和操作成本。

创建库表模板的详细步骤如下：

点击新建模板。从页面上方点击库表模板目录进入库表模块页面，点击创建模板按钮新建库表模板；
注意项:
用户创建的库表模板不能重名，库表字段的描述格式为（字段1名称，字段1类型，字段1描述:段2名称，字段2类型，字段2描述)
库表模板管理提供对库表的删除、修改和搜索操作。
库表模板创建完成后，用户可在新建实时计算任务填写库表信息时，直接从右上角点击导入，快速导入以保存的库表模板。

2.3 实战案例

前置条件：sql是使用的所有非临时表必须事先定义好，并且测试通过，否则在任务运行过程会有异常抛出。

2.3.1 案例：星级游戏实时指标统计

需求描述：

需求：实时统计多款星级游戏的注册人数。每款游戏的日志数据记录在各自的注册流水表。以1分钟为单位实时统计每款游戏的以上注册人数，并将结果数据写入到用户指定结果数据表。目前对以下4款游戏进行计算：御龙在天，英雄联盟。

实现方案：

实现思路：首先对每一款游戏的各个指标分别进行统计，然后将每款游戏相同的结果指标通过UNION的方式合并并存储结果表。完整的sql如下：

with(select "ylzt" bname, iUin qqid, dtEventTime agtime from ylzt_dsl_createrole_fht0) ylzt_cr, 
(select "lol" bname, iUin qqid, dtEventTime agtime from lol_dsl_createrole_fht0) lol_cr, 
(select "bns" bname, iUin qqid, dtEventTime agtime from bns_dsl_createrole_fht0) bns_cr, 
(select "nizhan" bname, iUin qqid, dtEventTime agtime from nizhan_dsl_createrole_fht0) nizhan_cr
insert into rtc_idata_createrole_m select from_unixtime(AGGRTIME DIV 1000, 'yyyyMMddHHmm') vdatetime_m, bname vgame, count(qqid) from (select bname, qqid, agtime from ylzt_cr union all select bname, qqid, agtime from lol_cr union all select bname, qqid, agtime from bns_cr union all select bname, qqid, agtime from nizhan_cr) utbl group by bname coordinate by agtime*1000 with aggr interval 60 seconds

在上述sql中，首先对每个表选出需要使用的字段：对于目前的需求，只用到iUin和dtEventTime字段，因此将这两个字段选出来，同理，对于收入流水表，选出收入相关字段。
要点：这段sql中使用了with语法，将一段子查询逻辑抽象出来作为临时表，以供后面的逻辑使用。
将每款游戏的数据UNION到一起，然后进行聚合统计。
首先从上一段sql中的临时表中把每个游戏的注册信息进行UNION，然后进行聚合统计计算，最后将数据写入目标表。Group by bname定义了聚合分组字段为bname，coordinate by定义了时间协调坐标，这里要求一个毫秒数，而agtime是源数据表中定义的dtEventTime字段是一个秒数，所以这里乘以1000。Aggr interval 60 seconds表示聚合粒度为1分钟。Count(qqid)以及sum(m)就是聚合统计。另外AGGRTIME是一个系统关键字，表示聚合时间的毫秒数，并且是aggr interval的整数倍，这个字段只有在当前查询（子查询）中包含group by时才有意义，否则使用这个关键字会报错。

2.3.2 案例：累加聚合和滑动窗口聚合

为了统一介绍累加聚合和滑动和滑动窗口聚合函数的使用，这里假定一个需求。

需求描述：

需求：假定某款业务需要按照如下规则进行实时统计：每1分钟进行一次数据统计输出当前分钟的登录人数，当前小时截止到当前分钟的登录人数，截止到当前分钟的连续30分钟的登录人数。

实现方案：

使用累加聚合和滑动窗口聚合两种方法来实现。完整的sql如下：

INSERT INTO dest SELECT appId, count(qq), count(qq ACCU), count(qq SW) FROM src GROUP BY appId  COORDINATE BY dTime WITH AGGR INTERVAL 60 SECONDS WITH ACCU INTERVAL 3600 SECONDS WITH SW INTERVAL 1800 SECONDS

说明：

在sql中通过WITH AGGR INTERVAL 60 SECONDS指定聚合窗口为1分钟，通过WITH ACCU INTERVAL 3600 SECONDS指定累加聚合窗口为1小时，通过WITH SW INTERVAL 1800 SECONDS指定滑动窗口为30分钟。 Count(qq)表示当前分钟的计数，count（qq ACCU)表示当前小时截止到当前分钟的计数，count(qq SW)表示当前分钟及向前30分钟的滑动窗口计数。

2.3.3 案例：关联维表实时指标统计

需求描述

实时计算微信在不同国家和地区，不同的设备的连接数和平均连接耗时。微信连接数据中，每条记录包含一个连接ip，连接设备类型（Android，ios），连接耗时等信息。
存在一张IP表，包含ip段到地区的映射关系（维表）。
要求按照1分钟的统计粒度计算每个国家每种设备的连接数和平均连接耗时。

实现方案

实现思路：对于每条数据首先针对ip通过维表关联的方式从ip表中读取维表中的国家和地区信息，然后按照国家和设备进行聚合统计。完整的sql如下：

INSERT INTO weixin_res_meta SELECT from_unixtime(AGGRTIME DIV 1000, 'yyyyMMddHHmm'), ttt.Device_, ip_table.country_id, 

COUNT(1), avg(CostTimeref_) FROM (SELECT Device_, CostTimeref_, CAST ((ClientIP_ - ClientIP_ % 256) AS STRING) AS ipstr 

FROM log_10410 WHERE Funid_='138') ttt LEFT JOIN ip_table ON ttt.ipstr = ip_table.ipint GROUP BY ttt.Device_, ip_table.country_id 

COORDINATE BY TimeStamp_*1000 WITH AGGR INTERVAL 60 SECONDS

Previous1. 数据接入 Next3. 工作流

Last updated 4 years ago

Was this helpful?