基于阿里云 EMR Serverless Spark 版快速搭建OSS日志分析应用

本文涉及的产品
EMR Serverless Spark 免费试用,1000 CU*H 有效期3个月
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 本文演示了使用 EMR Serverless Spark 产品搭建一个日志分析应用的全流程,包括数据开发和生产调度以及交互式查询等场景。

背景

随着互联网服务的广泛普及与技术应用的深入发展,日志数据作为记录系统活动、用户行为和业务操作的宝贵资源,其价值愈发凸显。然而,当前海量日志数据的产生速度已经远远超出了传统数据分析工具的处理能力,这不仅要求我们具备高效的数据收集和存储机制,更呼唤着强大、灵活且易用的数据分析平台的诞生。在此背景下,Apache Spark,这一专为大规模数据处理而设计的计算引擎,成为了构建高性能日志分析应用的理想选择。


阿里云 EMR Serverless Spark 是一款全托管、一站式的数据处理平台,基于Spark Native Engine构建,专为大规模数据处理和分析设计,提供弹性、高效的服务,让用户无需关注基础设施管理,100%兼容Spark,简化从开发到运维的全链路工作流程。


本文将以 OSS 日志处理场景为例,演示使用 EMR Serverless Spark 产品快速搭建日志分析应用。


OSS-HDFS 审计日志简介

阿里云的 OSS-HDFS 服务,是专为大数据处理和云原生数据湖存储设计的产品。该服务由阿里云的JindoFS提供技术支持,旨在无缝桥接阿里云对象存储(OSS)与 HDFS 生态系统,为 Apache Hadoop、Hive、Spark、Flink 等大数据处理框架提供高性能、高兼容性的存储解决方案。


在阿里云 OSS 控制台创建一个新的 OSS Bucket 时可以选择开通 HDFS 服务,创建完成后新的 OSS Bucket 即可支持 HDFS 接口访问:


HDFS审计日志(Audit Log)是Hadoop分布式文件系统(HDFS)的一个重要组成部分,它详尽地记录了所有用户对 HDFS 执行的操作信息。这些日志对于系统管理员监控、安全审计以及故障排查至关重要。每当用户通过 HDFS 的 NameNode 执行操作(如读取、写入、删除文件或目录等),NameNode 就会生成一条审计日志记录。类似于开源版 HDFS,OSS-HDFS 默认就支持 auditlog 日志,在根目录下的 /.sysinfo/auditlog 目录下保存了近一个月的审计日志,并且按照日期目录进行切分。


审计日志条目通常包含一些关键信息,比如操作时间、操作人、操作成功与否、来源IP、操作命令、操作目标文件等。下面三条日志分别记录了delete、getfileinfo和mkdir操作详情:

2024-05-14 00:12:37.746 allowed=true    ugi=hadoop (auth:UNKNOWN)       ip=172.16.0.99  cmd=delete      src=/tmp/hive/hadoop/c7e1564c-7f3f-4fe9-b993-1d56e1ddcd47       dst=<null>      perm=<null>     proto=rest
2024-05-14 00:07:36.652 allowed=true    ugi=hadoop (auth:UNKNOWN)       ip=172.16.0.99  cmd=getfileinfo src=/tmp/hive/hadoop/34567e8a-a5f2-4f6b-802c-ca4db2cc1d58       dst=<null>      perm=<null>     proto=rest
2024-05-14 00:12:37.316 allowed=true    ugi=hadoop (auth:UNKNOWN)       ip=172.16.0.99  cmd=mkdirs      src=/tmp/hive/hadoop/c7e1564c-7f3f-4fe9-b993-1d56e1ddcd47/_tmp_space.db dst=<null>      perm=hadoop:supergroup:rwx------        proto=rest


EMR Serverless Spark 工作空间简介

使用 EMR Serverless Spark 产品之前,需要了解工作空间相关的概念,工作空间是 EMR Serverless Spark 为业务开发划分的基本单元,是任务、资源和权限的集合。


接下去就可以参考产品的快速入门文档来体验:

  1. 阿里云账号角色授权:开通工作空间的前置操作
  2. 创建Spark工作空间:需要提前开通 OSS 和 DLF 等阿里云服务
  3. SQL任务快速入门:接下去的 EMR Serverless Spark 任务开发会使用 SQL 任务


EMR Serverless Spark 任务开发

下面我们来演示如何通过EMR Serverless Spark搭建一个日志分析应用。日志分析的一个很常见的需求是分析前一天访问 OSS-HDFS Bucket 的来源IP,比如希望找到有来自某些IP的异常突发流量,或者在事后调查敏感文件是否被异常IP所访问。


因为SQL是在数据分析中最常用的工具,所以使用 Spark SQL 来分析OSS-HDFS的审计日志。前面我们已经通过《SQL任务快速入门》对 SparkSQL 类的任务有了简单的了解,这部分内容会针对数据仓库源数据层、明细层、汇总层分别创建一个 SQL 任务。


源数据层

首先是日志文件的来源,我们要建立一个源数据层(ODS)的表,因为审计日志已经被归档到OSS-HDFS的系统目录里,所以我们可以通过Spark SQL建一个CSV外表:

  1. 表路径指向系统目录 oss://..oss-dls.aliyuncs.com/.sysinfo/auditlog/ (在操作的时候需要将 BUCKET_NAME 和 REGION_ID 替换为实际使用的 OSS Bucket 名称和所在地域)
  2. 日志条目中的不同字段用制表符(tab)分隔,所以指定 sep = '\t'
  3. 使ds是ServerlessSpark开发和调度平台使用的内置变量,代表业务日期(T−1)。比如在2024年5月21日运行的SQL任务,业务时间是指前一天,{ds} 是 Serverless Spark开发和调度平台使用的内置变量,代表业务日期(T-1)。比如在2024年5月21日运行的SQL任务,业务时间是指前一天,{ds}=2024-05-20(在这里无需手动替换 ds 值)
DROP TABLE IF EXISTS s_oss_hdfs_audit_tmp;
CREATE TABLE s_oss_hdfs_audit_tmp (
  tm string,
  allowed string,
  ugi string,
  ip string,
  cmd string,
  src string,
  dst string,
  perm string,
  proto string
) USING csv 
OPTIONS (
  path 'oss://<BUCKET_NAME>.<REGION_ID>.oss-dls.aliyuncs.com/.sysinfo/auditlog/${ds}',
  sep '\t'
);


把这个SQL文件(s_oss_hdfs_audit_tmp.sql)保存后,点击发布。


数仓明细层

其次,我们要基于这个ODS外表创建一张数据仓库明细层(DWD)表,以Parquet格式存储,并按天进行分区。我们需要对 ODS 表进行简单的清晰和转换,比如把access_time从字符串转换timestamp成类型,将字段内容 ip=172.16.0.99 转换为 IP 地址 172.16.0.99 等。这个SQL里同样使用了 ${ds} 内置变量。

CREATE TABLE IF NOT EXISTS dwd_oss_hdfs_audit_di (
  access_time timestamp,
  allowed string,
  ugi string,
  ip string,
  cmd string,
  src string,
  dst string,
  perm string,
  proto string,  
  dt string
) 
USING parquet 
PARTITIONED BY (dt);
INSERT OVERWRITE TABLE dwd_oss_hdfs_audit_di partition (dt = '${ds}')
SELECT 
  to_timestamp(tm),
  split_part(allowed, '=', 2),
  split_part(ugi, '=', 2),
  split_part(ip, '=', 2),
  split_part(cmd, '=', 2),
  split_part(src, '=', 2),
  split_part(dst, '=', 2),
  split_part(perm, '=', 2),
  split_part(proto, '=', 2)
FROM
  s_oss_hdfs_audit_tmp;


把这个SQL文件(dwd_oss_hdfs_audit_di.sql)保存后,点击发布。


数仓汇总层

最后,我们对数仓明细层数据做一个简单的分析,取出前一天请求量最大的20个IP地址,我们会创建一张 DWS 汇总表:

CREATE TABLE IF NOT EXISTS dws_oss_hdfs_ip_ana (
  ip string,
  cnt bigint,
  dt string
) 
USING parquet 
PARTITIONED BY (dt);
INSERT OVERWRITE TABLE dws_oss_hdfs_ip_ana partition (dt = '${ds}')
SELECT 
  ip,
  count(*) cnt
FROM
  dwd_oss_hdfs_audit_di
WHERE
  dt = '${ds}'
GROUP BY ip
ORDER BY cnt DESC
LIMIT 20;


把这个SQL文件(dws_oss_hdfs_ip_ana.sql)保存后,点击发布。


EMR Serverless Spark 任务编排

创建工作流

在前面的章节中,我们已经分别在数据仓库源数据层、明细层、汇总层各创建一个 SQL 任务,这些任务都处于“已发布”状态。接下去我们需要创建一个工作流把这三个SQL任务进行适当的编排,并且让工作流能在每天的固定时间进行调度。


在 Serverless Spark 工作空间的导航栏中找到“任务编排”链接,点击“创建工作流”后进入新建工作流 oss_hdfs_auditlog 的配置界面。在这个界面里需要填写工作流名称和资源队列,同时可以选择调度类型是“调度器”,调度周期是每天的 00:05。



编辑节点

在编辑工作流的页面,鼠标左键双击节点,或者单击下方的添加节点,进入节点编辑页面。我们需要按顺序选择s_oss_hdfs_audit_tmp、dwd_oss_hdfs_audit_di、dws_oss_hdfs_ip_ana节点,加入到工作流中。


同时也需要配置节点依赖关系,比如 dwd_oss_hdfs_audit_di 节点的上游节点是 s_oss_hdfs_audit_tmp,dws_oss_hdfs_ip_ana 节点的上游节点是 dwd_oss_hdfs_audit_di。


三个节点编辑完成之后,自动生成如下 DAG,完成工作流的编辑。


发布工作流

在工作流编辑页面右上角,点击“发布工作流”,在输入发布信息后点击“确认”,完成工作流的发布。


发布工作流之后自动跳转回到工作流列表,我们可以看到新创建的工作流。打开“调度状态”开关,之后工作流会根据调度器的设置进行按天调度。


点击工作流名称,进入工作流调度实例列表,在这里可以看到每次调度运行的成功或失败的任务节点,也可以点击右上角的“手动运行”按钮进行一次手动调度。


在每天凌晨的定时调度完成之后或者一次手动调度成功之后,我们可以回到 SQL 任务开发界面,在编辑器中输入如下 SQL 查询语句,可以快速获取到前一天请求 OSS-HDFS 数量最多的前 20 个 IP 地址:

SELECT * FROM dws_oss_hdfs_ip_ana where dt = '${ds}';


我们可以得到类似下面的 SQL 运行结果:


总结

本文演示了使用 EMR Serverless Spark 产品搭建一个日志分析应用的全流程,包括数据开发和生产调度以及交互式查询等场景。


EMR Serverless Spark 在 2024年5月正式开启公测,在公测期间可以免费使用最高 100 CU 计算资源,欢迎试用。如果您在使用 EMR Serverless Spark 版的过程中遇到任何疑问,可钉钉扫描以下二维码加入钉钉群(群号:58570004119)咨询。

快速跳转

  1. EMR Serverless Spark 版官网:https://www.aliyun.com/product/bigdata/serverlessspark
  2. 产品控制台:https://emr-next.console.aliyun.com/
  3. 产品文档:https://help.aliyun.com/zh/emr/emr-serverless-spark/
  4. SQL 任务快速入门:https://help.aliyun.com/zh/emr/emr-serverless-spark/getting-started/get-started-with-sql-task-development
相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
目录
打赏
0
7
7
0
1227
分享
相关文章
基于阿里云 EMR Serverless Spark 版快速搭建OSS日志分析应用
基于阿里云 EMR Serverless Spark 版快速搭建OSS日志分析应用
EMR Serverless StarRocks 全面升级:重新定义实时湖仓分析
本文介绍了EMR Serverless StarRocks的发展路径及其架构演进。首先回顾了Serverless Spark在EMR中的发展,并指出2021年9月StarRocks开源后,OLAP引擎迅速向其靠拢。随后,EMR引入StarRocks并推出全托管产品,至2023年8月商业化,已有500家客户使用,覆盖20多个行业。 文章重点阐述了EMR Serverless StarRocks 1.0的存算一体架构,包括健康诊断、SQL调优和物化视图等核心功能。接着分析了存算一体架构的挑战,如湖访问不优雅、资源隔离不足及冷热数据分层困难等。
基于OpenLake的Flink+Paimon+EMR StarRocks流式湖仓分析
阿里云OpenLake解决方案建立在开放可控的OpenLake湖仓之上,提供大数据搜索与AI一体化服务。通过元数据管理平台DLF管理结构化、半结构化和非结构化数据,提供湖仓数据表和文件的安全访问及IO加速,并支持大数据、搜索和AI多引擎对接。本文为您介绍以Flink作为Openlake方案的核心计算引擎,通过流式数据湖仓Paimon(使用DLF 2.0存储)和EMR StarRocks搭建流式湖仓。
731 5
基于OpenLake的Flink+Paimon+EMR StarRocks流式湖仓分析
阿里云MaxCompute-XGBoost on Spark 极限梯度提升算法的分布式训练与模型持久化oss的实现与代码浅析
本文介绍了XGBoost在MaxCompute+OSS架构下模型持久化遇到的问题及其解决方案。首先简要介绍了XGBoost的特点和应用场景,随后详细描述了客户在将XGBoost on Spark任务从HDFS迁移到OSS时遇到的异常情况。通过分析异常堆栈和源代码,发现使用的`nativeBooster.saveModel`方法不支持OSS路径,而使用`write.overwrite().save`方法则能成功保存模型。最后提供了完整的Scala代码示例、Maven配置和提交命令,帮助用户顺利迁移模型存储路径。
EMR Serverless Spark:一站式全托管湖仓分析利器
本文根据2024云栖大会阿里云 EMR 团队负责人李钰(绝顶) 演讲实录整理而成
359 2
Serverless MCP 运行时业界首发,函数计算让 AI 应用最后一公里提速
作为云上托管 MCP 服务的最佳运行时,函数计算 FC 为阿里云百炼 MCP 提供弹性调用能力,用户只需提交 npx 命令即可“零改造”将开源 MCP Server 部署到云上,函数计算 FC 会准备好计算资源,并以弹性、可靠的方式运行 MCP 服务,按实际调用时长和次数计费,欢迎你在阿里云百炼和函数计算 FC 上体验 MCP 服务。
169 29
云大使 X 函数计算 FC 专属活动上线!享返佣,一键打造 AI 应用
如今,AI 技术已经成为推动业务创新和增长的重要力量。但对于许多企业和开发者来说,如何高效、便捷地部署和管理 AI 应用仍然是一个挑战。阿里云函数计算 FC 以其免运维的特点,大大降低了 AI 应用部署的复杂性。用户无需担心底层资源的管理和运维问题,可以专注于应用的创新和开发,并且用户可以通过一键部署功能,迅速将 AI 大模型部署到云端,实现快速上线和迭代。函数计算目前推出了多种规格的云资源优惠套餐,用户可以根据实际需求灵活选择。
鹰角网络:EMR Serverless Spark 在《明日方舟》游戏业务的应用
鹰角网络为应对游戏业务高频活动带来的数据潮汐、资源弹性及稳定性需求,采用阿里云 EMR Serverless Spark 替代原有架构。迁移后实现研发效率提升,支持业务快速发展、计算效率提升,增强SLA保障,稳定性提升,降低运维成本,并支撑全球化数据架构部署。
鹰角网络:EMR Serverless Spark 在《明日方舟》游戏业务的应用
Serverless MCP 运行时业界首发,函数计算让 AI 应用最后一公里提速
Serverless MCP 运行时业界首发,函数计算支持阿里云百炼 MCP 服务!阿里云百炼发布业界首个全生命周期 MCP 服务,无需用户管理资源、开发部署、工程运维等工作,5 分钟即可快速搭建一个连接 MCP 服务的 Agent(智能体)。作为云上托管 MCP 服务的最佳运行时,函数计算 FC 为阿里云百炼 MCP 提供弹性调用能力。
163 0
 Serverless MCP 运行时业界首发,函数计算让 AI 应用最后一公里提速
Serverless + AI 让应用开发更简单,加速应用智能化
Serverless + AI 让应用开发更简单,加速应用智能化