概述
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 #停留时间。
AI 代码解读
上述信息在APP的日志中使用逗号分隔符，所以在LogHub配置指定采集模式时选择逗号分隔。
SparkStreaming 对接APP
SparkStreaming 对接APP可以使用X-Pack Connectors中对接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)
);
AI 代码解读
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()
}
}
AI 代码解读
SparkStreaming同步数据到Phoenix后，可以对Phoenix数据库进行用户明细查询。例如：
# 查询用户user_id_1006所有浏览明细。
select * from user_event where user_id = 'user_id_1006';
AI 代码解读
同步到Spark离线数仓
Phoenix在线数据库适合明细查询，如果需要进行统计、离线计算需要用到Spark数仓。Phoenix同步数据到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
AI 代码解读
离线数仓批量计算
数据同步到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
AI 代码解读
计算的结果可以回写到业务数据库，供业务查询、出报表等。
小结
本文简单介绍了Spark如何对接LogHub以及如何同步数据等常用的操作。参考链接如下：
- X-Pack Spark介绍参考：X-Pack Spark基本介绍。
