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

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

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

前言


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

image

场景设计


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

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

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

技术选型


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

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

预备工作


创建EMR集群

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

开通ECS和日志服务

详细步骤请参考官方文档。
其中ilogtail配置如下
image
日志收集到LogHub以后,可以通过官网控制台查看。本文示例中收集到的数据如下:
image

创建Tablestore结果表

image
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文件。
image

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; 查看表结构如下。
image

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中的结果数据如下:

image

性能调优


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

数据读取

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

设置合理的批处理时间

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

增加作业资源

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

数据写入

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

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
17天前
|
存储 分布式计算 资源调度
通过日志聚合将作业日志存储在HDFS中
如何通过配置Hadoop的日志聚合功能,将作业日志存储在HDFS中以实现长期保留,并详细说明了相关配置参数和访问日志的方法。
15 0
通过日志聚合将作业日志存储在HDFS中
|
6天前
|
分布式计算 Java Apache
Apache Spark Streaming技术深度解析
【9月更文挑战第4天】Apache Spark Streaming是Apache Spark生态系统中用于处理实时数据流的一个重要组件。它将输入数据分成小批次(micro-batch),然后利用Spark的批处理引擎进行处理,从而结合了批处理和流处理的优点。这种处理方式使得Spark Streaming既能够保持高吞吐量,又能够处理实时数据流。
23 0
|
1月前
|
SQL 存储 分布式计算
|
1月前
|
分布式计算 Apache 数据安全/隐私保护
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
38 1
|
1月前
|
应用服务中间件 nginx Docker
[loki]轻量级日志聚合系统loki快速入门
[loki]轻量级日志聚合系统loki快速入门
|
1月前
|
分布式计算 资源调度 测试技术
“Spark Streaming异常处理秘籍:揭秘如何驯服实时数据流的猛兽,守护你的应用稳如泰山,不容错过!”
【8月更文挑战第7天】Spark Streaming 是 Apache Spark 中的关键组件,用于实时数据流处理。部署时可能遭遇数据问题、资源限制或逻辑错误等异常。合理处理这些异常对于保持应用稳定性至关重要。基础在于理解其异常处理机制,通过 DSC 将数据流切分为 RDD。对于数据异常,可采用 try-catch 结构捕获并处理;资源层面异常需优化 Spark 配置,如调整内存分配;逻辑异常则需加强单元测试及集成测试。结合监控工具,可全面提升应用的健壮性和可靠性。
58 3
|
3月前
|
分布式计算 Java Scala
如何处理 Spark Streaming 的异常情况?
【6月更文挑战第16天】如何处理 Spark Streaming 的异常情况?
172 56
|
2月前
|
存储 并行计算 开发工具
SLS Prometheus存储问题之相比客户端SDK聚合写入,SLS网关侧聚合写入有什么优势
SLS Prometheus存储问题之相比客户端SDK聚合写入,SLS网关侧聚合写入有什么优势
|
2月前
|
分布式计算 监控 数据处理
Spark Streaming:解锁实时数据处理的力量
【7月更文挑战第15天】Spark Streaming作为Spark框架的一个重要组成部分,为实时数据处理提供了高效、可扩展的解决方案。通过其微批处理的工作模式和强大的集成性、容错性特性,Spark Streaming能够轻松应对各种复杂的实时数据处理场景。然而,在实际应用中,我们还需要根据具体需求和资源情况进行合理的部署和优化,以确保系统的稳定性和高效性。
|
2月前
|
SQL 分布式计算 DataWorks
DataWorks产品使用合集之怎么编写和执行Spark SQL
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。