海量监控日志基于EMR Spark Streaming SQL进行实时聚合

作者:伯箫,阿里云高级开发工程师。现在在阿里云表格存储团队,负责管控系统的开发,对NOSQL类数据库系统有一些了解。

前言


从EMR-3.21.0 版本开始将提供Spark Streaming SQL的预览版功能,支持使用SQL来开发流式分析作业。结果数据可以实时写入Tablestore。
本文以LogHub为数据源,收集ECS上的日志数据,通过Spark Streaming SQL进行聚合后,将流计算结果数据实时写入Tablestore,展示一个简单的日志监控场景。

海量监控日志基于EMR Spark Streaming SQL进行实时聚合

场景设计


假设有一个商品表Goods,商品信息开放给用户浏览,用户浏览完以后会产生以下格式的日志数据:

"RequestId":"c85df119-f6db-449f-89bb-6773d2468f89",
"Time":2019-07-30 12:05:28,
"

我们需要将原始日志数据,根据GoodsName、OperationType和时间,聚合成一分钟一个点的监控数据。用于监控各商品的访问情况。
最终需要的监控数据格式如下:
海量监控日志基于EMR Spark Streaming SQL进行实时聚合
过程如下图所示,比较简单。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

技术选型


本文主要来看一下结果数据存放数据库的选型。对于本文的监控场景,结果数据的量级,取决于商品数量。真实情况下,可能还要增加商品的规格、颜色等监控指标,结果数据量会比较大。此外,对于时间粒度比较小的监控数据,一般都只需要保留最近的,时间比较久的历史数据需要删除。
将Tablestore与传统关系型数据库MySQL进行对比,Tablestore有以下优势:

  • 支持海量数据,无缝扩展
    表格存储通过数据分片和负载均衡技术,实现了无缝扩展。
  • 支持数据自动过期
    数据生命周期(Time To Live,简称 TTL)是数据表的一个属性,即数据的存活时间,单位为秒。表格存储会在后台对超过存活时间的数据进行清理,以减少用户的数据存储空间,降低存储成本。监控场景下比较适用,不需要手动去删除数据。

预备工作


创建EMR集群

开通EMR之前,先要对云账号进行实名认证,然后创建默认的EMR角色并授予 AliyunEMRDefaultRole和AliyunEmrEcsDefaultRole这两个角色。
EMR暂时还不支持在官网控制台上配置写入Tablestore的任务,需要登录到MER集群的机器上去操作,所以开通EMR集群的时候,请选择自定义购买,并在最后一步打开挂载公网、远程登录、密码方式登录,自己设置一个密码,如下图。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合
开通完以后,进入到ECS控制台,会看到有一台Master节点机器有弹性IP。后面步骤中,操作EMR Spark Streaming需要使用这个IP地址远程登录到机器上。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

开通ECS和日志服务

详细步骤请参考官方文档。
其中ilogtail配置如下
海量监控日志基于EMR Spark Streaming SQL进行实时聚合
日志收集到LogHub以后,可以通过官网控制台查看。本文示例中收集到的数据如下:
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

创建Tablestore结果表

海量监控日志基于EMR Spark Streaming SQL进行实时聚合
Count列,作为属性列,不需要定义在主键中。

数据处理流程


1、下载支持数据源需要的jar包

下载地址
https://github.com/aliyun/aliyun-emapreduce-sdk/blob/master-2.x/jars/datasources/latest/emr-datasources_shaded_2.11-1.7.0.jar

2、进入streaming-sql执行环境

下载完jar包以下,在包所在目录执行以下命令进入交互式开发环境。

注意:jar包要上传到集群机器上,不支持远程引用oss文件。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

3、创建LogHub数据源表

创建LogHub数据源表之前,需要手动开通日志服务的project和logStore,并将日志数据收集好,具体参考【预备工作】一节中,开通LogHub的官方文档。
然后再到EMR集群机器的交互式执行环境中创建LogHub数据源表,示例如下:

CREATE TABLE loghub_source(GoodsName string,OperationType string,RequestId string,__time__ timestamp)
USING loghub
OPTIONS (
sls.project = '{your project name}',
sls.store = '{store name}',
access.key.id = '{access-key}',
access.key.secret = '{access-key-secret}',
endpoint = 'http://{your project name}.cn-hangzhou-intranet.log.aliyuncs.com');

上面的sql语句,创建一个database为helloemr,tablename为loghub_source的表,shema中有GoodsName、OperationType、Time、RequestId、__time__五个字段,其中__time__字段是在配置ilogtail的时候解析日志中的Time字段得到的,等同于Time字段。
需要注意的是,Endpoint请使用内网域名,公网域名速度上会慢很多。

4、创建Tablestore结果表

同样的,创建Tablestore结果表之前,也要先到官网控制台建到tablestore的实例和表。
创建EMR的Tablestore结果表,示例如下

CREATE TABLE tablestore_sink
USING tablestore
OPTIONS(
endpoint="https://sparkStreaming.cn-hangzhou.vpc.tablestore.aliyuncs.com",
access.key.id="{access-key}",
access.key.secret="{access-key-secret}",
table.name="tablestore_sink",
instance.name="sparkStreaming",
catalog='{"columns":{"GoodsName":{"col":"GoodsName","type":"string"},"OperationType":{"col":"OperationType","type":"string"},"Time":{"col":"Time","type":"long"},"Count":{"col":"Count","type":"long"}}}');

以上sql代码,创建了一个表名为tablestore_sink的表,实例名是sparkStreaming。注意:Endpoint请使用vpc域名。
创建成功以后,使用desc tablestore_sink; 查看表结构如下。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

5、结果数据写入到Tablestore

通过GoodsName、OperationType聚合一分钟内的请求次数。这里要用到Spark Streaming的滚动窗口函数,取window.start作为聚合后的时间。

SET spark.sql.streaming.checkpointLocation.loghub_source=/home/helloemr;
SET spark.sql.streaming.query.trigger.loghub_source=ProcessingTime;
SET spark.sql.streaming.query.trigger.intervalMs.loghub_source=10000;
INSERT INTO tablestore_sink
SELECT GoodsName,OperationType,count(*) as Count,to_unix_timestamp(window.start, 'yyyy-MM-dd HH:mm:ss') as Time from loghub_source
where delay(__time__)<"2 minute" 
GROUP BY TUMBLING (__time__, interval 1 minute),GoodsName,OperationType;

示例SQL中只填了几个必填参数,具体可以参考作业模板。其中checkpointLocation代表本次流式查询作业的checkpoint路径,需要设置一个绝对路径值。
最终,Tablestore中的结果数据如下:

海量监控日志基于EMR Spark Streaming SQL进行实时聚合

性能调优


Spark Streaming 是基于Spark的流式处理引擎,其基本原理是把输入数据以某一时间间隔批量的处理,当批处理间隔缩短到秒级时,便可以用于处理实时数据流。它本质上是微批处理。
使用上一节的示例代码,实际测试下来,一次作业需要耗费1秒左右的时间。在实际实例中,由于源表、目标数据表、数据量大小以及Sql的复杂程度不同,耗费的时间也会不同。

数据读取

本文使用LogHub为数据源,Shard数量多,Spark Streaming的作业并发度也会多,但需要设置合理的Shard数,具体请参考日志服务分区设置。
同时,创建LogHub数据源表的时候,请使用内网Endpoint。
Spark Streaming作业调优

设置合理的批处理时间

trigger.intervalMs代表批次间隔,单位毫秒,默认为0L。运行任务的时候,当间隔时间比一次任务的运行时间短的时候,任务会打印WARN日志。一般这个值的大小如果能够使得Streaming作业刚好处理好上一个的批处理的数据,那么这个就是最优值。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

增加作业资源

EMR官网控制台里面有的监控大盘,可以看到作业占用的资源情况,可以根据实际情况调整分配给作业的资源大小。

数据写入

对于表格存储(Tablestore)来说,合理的主键设计,是提高写入性能的关键因素。具体可以参考表格存储最佳实践。一个设计良好的主键,需要避免访问压力集中在一个小范围的连续的分片键上,也就是说避免热点分片。设计良好的表结构,整张表的访问压力能够均匀的分散在各个分片上,这样才能充分利用后端服务器的能力。
如果是新建的Tablestore表,而且数据写入量比较大,最好联系下@表格存储技术支持,对表进行预分区。可以提高初始状态下的写入性能。
海量监控日志基于EMR Spark Streaming SQL进行实时聚合

上一篇:Databricks数据洞察 限时免费开启公测!


下一篇:一文解开java中字符串编码的小秘密