X-Pack Spark对接阿里云日志服务LogHub

简介: 概述 X-Pack Spark分析引擎是基于Spark提供的复杂分析、流式处理、机器学习的能力。Spark分析引擎可以对接阿里云的多种数据源,例如:云HBase数据库、MongoDB、Phoenix等,同时也支持对接阿里云日志服务LogHub。

概述

X-Pack Spark分析引擎是基于Spark提供的复杂分析、流式处理、机器学习的能力。Spark分析引擎可以对接阿里云的多种数据源,例如:云HBase数据库、MongoDB、Phoenix等,同时也支持对接阿里云日志服务LogHub。阿里云日志服务(Log Service,简称LOG)是针对实时日志数据的一站式服务,提供日志类数据采集、消费、投递及查询分析功能,全面提升海量日志处理和分析能力。

场景介绍

某一款销售平台的APP,针对用户在APP中打开首页、搜索、商品详细页以及最终下订单购买商品等操作,操作所产生的事件均记录到阿里云日志系统中。现需要对APP的用户的行为数据做一些统计分析,每天、每周出详细的运营数据、以及给用户提供在线查询账单等。

如何实现

通过阿里云的日志服务+X-Pack Spark+云HBase完成这些诉求。先看下整理的数据流图:

_


由上图可见数据流程为:用户通过LogHub对接APP的日志数->Spark Streming 对接LogHub同步数到HBase(Phoenix)->在线数据同步到Spark离线数仓->离线数仓批量计算输出运营数据等。
APP日中包含用户的使用APP所产生的事件信息,下面以一个简单的例子说明下每一个步骤的实现。

LogHub对接APP日志

假设APP的日志产生在某机器的目录文件中,通过LogHub可以对接机器的文件,读取解析日志。假设日志的字段信息如下:

event_time: long #事件产生的时间戳
user_id: string #用户ID,唯一值。
device_id: String #设备id,APP使用的设备。
event_name: String #事件名称,如:首页、搜索、明细页、购买
prod_id: String #商品ID。
stay_times: int #停留时间。

上述信息在APP的日志中使用逗号分隔符,所以在LogHub配置指定采集模式时选择逗号分隔。

SparkStreaming 对接APP

SparkStreaming 对接APP可以使用X-Pack Connectors中对接LogHub的插件(可参考:Spark对接LogHub快速入门)。SparkStreaming对接LogHub可以设置每个1分钟同步一次数据到Phoenix。
同步数据之前需要在Phoenix中创建一张表,如下:

CREATE TABLE IF NOT EXISTS user_event (
   event_time BIGINT NOT NULL,
   user_id VARCHAR NOT NULL,
   device_id VARCHAR,
   event_name VARCHAR,
   prod_id VARCHAR
   CONSTRAINT my_pk PRIMARY KEY (event_time, user_id)
  );

Phoenix表user_event使用user_event和user_id作为组合主键,主要是为了使用user_id进行运营明细查询,时间信息方便按照时间范围同步数据到Spark。
SparkStreaming同步LogHub数据到Phoenix的代码主要逻辑如下:

val loghubStream = LoghubUtils.createStream(
        ssc,
        loghubProject,
        logStore,
        loghubGroupName,
        endpoint,
        numReceiver,
        accessKeyId,
        accessKeySecret,
        StorageLevel.MEMORY_AND_DISK)

      loghubStream.foreachRDD { rdd =>
        rdd.foreachPartition { pt =>
          // 获取Phoenix的链接
          val phoenixConn = DriverManager.getConnection("jdbc:phoenix:" + zkAddress)
          val statment = phoenixConn.createStatement()
          var i = 0
          while (pt.hasNext) {
            val value = pt.next()
            //获取的LogHub的数据是json格式的,需要进行转换
            val valueFormatted = JSON.parseObject(new String(value))
            //构造phonenix 插入语句
            val insetSql = s"upsert into $phoenixTableName values(" +
              s"${valueFormatted.getLong("event_time")}," +
              s"'${valueFormatted.getString("user_id").trim}'," +
              s"'${valueFormatted.getString("device_id").trim}'," +
              s"'${valueFormatted.getString("event_name").trim}'," +
              s"'${valueFormatted.getString("prod_id").trim}')"
            statment.execute(insetSql)
            i = i + 1
            // 每隔batchSize行提交一次commit到Phoenix。
            if (i % batchSize == 0) {
              phoenixConn.commit()
              println(s"====finish upsert $i rows====")
            }
          }
          phoenixConn.commit()
          println(s"==last==finish upsert $i rows====")
          phoenixConn.close()
          }
      }

SparkStreaming同步数据到Phoenix后,可以对Phoenix数据库进行用户明细查询。例如:

# 查询用户user_id_1006所有浏览明细。
select * from user_event where user_id = 'user_id_1006';

同步到Spark离线数仓

Phoenix在线数据库适合明细查询,如果需要进行统计、离线计算需要用到Spark数仓。Phoenix同步数据到Spark数仓实质就是在Spark上创建表,然后把数据同步一份到Spark表中。
归档数据到Spark请参考:批量归档。 本文用Sql表示下同步的逻辑,这里假设数据每天同步一次到Spark 。
Spark 中建表、同步的方法如下:

#在Spark中创建Parquet格式表:user_event_parquet,使用dt作为分区字段。
create table user_event_parquet(
    event_time long,
    user_id string,
    device_id string,
    event_name string,
    prod_id string, 
    dt string
) using parquet
partitioned by(dt);

#  在Spark中创建表user_event_phoenix映射Phoenix数据库的表。
CREATE TABLE user_event_phoenix USING org.apache.phoenix.spark
OPTIONS (
'zkUrl' 'hb-xx-master3-001.hbase.rds.aliyuncs.com:2181,hb-xx-master1-001.hbase.rds.aliyuncs.com:2181,hb-xx-master2-001.hbase.rds.aliyuncs.com:2181',
'table' 'user_event'
);

# 向Parquet表:user_event_parquet插入一天:2019-01-01的数据
insert into user_event_parquet select EVENT_TIME,USER_ID,DEVICE_ID,EVENT_NAME,PROD_ID,'2019-01-01' from user_event_phoenix where EVENT_TIME >=1546272000 and EVENT_TIME < 1546358400

离线数仓批量计算

数据同步到Spark可以对Spark数据做统计分析预算,例如:

#统计每天的访问数
select dt, count(*) from user_event_parquet group by dt
#统计前十的访问
select dt, count(*) total from user_event_parquet group by dt order by total desc limit 10
#统计前100个用户的访问数
select dt,user_id, count(*) total  from user_event_parquet group by dt,user_id order by total desc limit 100

计算的结果可以回写到业务数据库,供业务查询、出报表等。

小结

本文简单介绍了Spark如何对接LogHub以及如何同步数据等常用的操作。参考链接如下:

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
目录
相关文章
|
4月前
|
分布式计算 监控 Spark
Spark 任务运行时日志分析
Spark 任务运行时日志分析
46 0
|
5月前
|
存储 数据可视化 搜索推荐
Kratos微服务轻松对接EFK日志系统
EFK 是一个完整的分布式日志收集系统,很好地解决了上述提到的日志收集难,检索和分析难的问题。EFK=Elasticsearch+Fluentd+Kibana
88 0
|
6月前
|
分布式计算 安全 Java
SPARK 应用如何快速应对 LOG4J 的系列安全漏洞
SPARK 应用如何快速应对 LOG4J 的系列安全漏洞
|
7月前
|
存储 消息中间件 监控
阿里云sls日志服务简介和使用流程
阿里云SLS(Simple Log Service)是一种高度可扩展的、低成本的日志托管服务,它提供了全面的日志采集、存储、分析和呈现功能。阿里云SLS是全球首个在公共云上提供日志服务的企业,它具有高可靠性、高稳定性和高安全性等特点,可满足不同企业的日志需求。
|
7月前
|
消息中间件 分布式计算 大数据
大数据Spark实时搜索日志实时分析
大数据Spark实时搜索日志实时分析
135 1
|
24天前
|
分布式计算 运维 大数据
阿里云 EMR Serverless Spark 版免费邀测中
阿里云 EMR Serverless Spark 版,以 Spark Native Engine 为基础,旨在提供一个全托管、一站式的数据开发平台。诚邀您参与 EMR Serverless Spark 版免费测试,体验 100% 兼容 Spark 的 Serverless 服务:https://survey.aliyun.com/apps/zhiliao/iscizrF54
377 0
阿里云 EMR Serverless Spark 版免费邀测中
|
7月前
|
分布式计算 自然语言处理 监控
大数据Spark对SogouQ日志分析
大数据Spark对SogouQ日志分析
189 0
|
3月前
|
存储 监控 Serverless
在处理阿里云函数计算3.0版本的函数时,如果遇到报错但没有日志信息的情况
在处理阿里云函数计算3.0版本的函数时,如果遇到报错但没有日志信息的情况【1月更文挑战第23天】【1月更文挑战第114篇】
63 5
|
3月前
|
存储 监控 数据挖掘
阿里云函数计算的日志可以通过两种方式进行处理
【1月更文挑战第16天】【1月更文挑战第76篇】阿里云函数计算(Function Compute)的日志可以通过两种方式进行处理
55 1
|
4月前
|
分布式计算 监控 分布式数据库
Spark Day05:Spark Core之Sougou日志分析、外部数据源和共享变量
Spark Day05:Spark Core之Sougou日志分析、外部数据源和共享变量
50 0