【最佳实践】基于阿里云数加·StreamCompute快速构建网站日志实时分析大屏

阿里云云栖号

2017-01-15

导读：本文偏向于实操层面，为大家介绍如何基于阿里云数加StreamCompute、DataV快速构建网站日志实时分析。

点击上方“云栖社区”可以订阅哦

本文偏向于实操层面，为大家介绍如何基于阿里云数加StreamCompute、DataV快速构建网站日志实时分析。

【使用前须知】

为保障本教程的顺利的进行，须知晓如下使用前提：

1.具备阿里云账号（淘宝及1688帐号可直接使用会员名登录）；

2.下载并安装Logstash的DataHub Output插件。

3.开通DataHub/StreamCompute/AnalyticDB或RDS/DataV产品；

【实现的业务场景】

数据来源于某网站上的HTTP访问日志数据，基于这份网站日志来实现如下分析需求：

1.实时统计并展现网站的PV和UV，并能够按照用户的终端类型（如Android、iPad、iPhone、PC等）分别统计。

2.实时统计并展现网站的流量来源。

3.从IP中解析出region或者经纬度在地图上进行展示。

【说明】浏览次数（PV）和独立访客（UV）是衡量网站流量的两项最基本指标。用户每打开一个网站页面，记录一个PV，多次打开同一页面PV 累计多次。独立访客是指一天内，访问网站的不重复用户数，一天内同一访客多次访问网站只计算1 次。Referer 可以分析网站访问来源，它是网站广告投放评估的重要指标，还可以用于分析用户偏好等。

【操作流程概述】

如上图所示，红色箭线部分为流式数据处理部分，主要拆解如下：

lÂ 配置Logstash，将网站产生的日志实时采集至DataHub。

lÂ 申请开通DataHub，创建项目Project及Topic（DataHub服务订阅和发布的最小单位）。

lÂ 开通StreamCompute，创建项目Project及注册数据输入源（DataHub）和输出源（RDS），并创建流任务（Stream SQL任务）。

lÂ 上一步骤中关于输出源RDS的配置，需要事先购买RDS for Mysql资源。

lÂ 申请开通DataV，新建RDS数据源并创建DataV项目进入大屏制作。

【数据结构设计】

DataHub Topic：分别创建Topic为：coolshell_log_tracker、coolshell_log_detail、coolshell_log_fact。

RDS：分别创建Table为： adm_refer_info、 adm_user_measures、 flyingline_coordinatesÂ ã

【网站日志实时解析】

Logstash安装与配置

配置前须知

阿里云流计算为了方便用户将更多数据采集进入DataHub，提供了针对Logstash的DataHub Output插件。

Logstash安装要求JRE 7版本及以上，否则部分工具无法使用。

操作步骤

步骤1 点击下载Logstash 2.4.1，点击下载。

步骤2 通过如下命令解压即可使用：

$ tar -xzvf logstash-2.4.1.tar.gz

$ cd logstash-2.4.1

步骤3 下载DataHub Logstash Output插件并使用如下命令进行安装：

$ {LOG_STASH_HOME}/bin/plugin install --local logstash-output-datahub-1.0.0.gem

步骤4 下载GeoIP解析IP数据库到本地。

wget http://geolite.maxmind.com/download/geoip/database/GeoLiteCity.dat.gz

步骤5 解压到当前路径并移动到Logstash具体路径下。

步骤6 配置Logstash任务.conf，示例如下:

input {

file {

path => "/Users/yangyi/logstash-2.4.1/sample/coolshell_log.log"

start_position => "beginning"

}

}

filter{

grok {

match => {

"message" => "(? [^ ]*) - (? [- ]*) \[(? [^\])*]\] \"(? \S+)(?: +(?< url>[^\"]*?)(?: +\S*)?)?(?: +(? [^\"]*))\" (? [^ ]*) (? [^ ]*) \"(? [^\"]*)\" \"(? [^\"]*)\""

}

}

geoip {

source => "ip"

fields => ["city_name","latitude", "longitude"]

target => "geoip"

database => "/Users/yangyi/logstash-2.4.1/bin/GeoLiteCity.dat"

add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]

add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]

}

mutate {

add_field=>{"region" => "%{[geoip][city_name]}"}

add_field=>{"coordinates" => "%{[geoip][coordinates]}"}

add_field=>{"x" => "%{[geoip][longitude]}"}

add_field=>{"y" => "%{[geoip][latitude]}"}

convert => [ "x", "float" ]

convert => [ "y", "float" ]

#convert => [ "coordinates", "float" ]

}

ruby{

code => "

md = event.get('accesstime')

event.set('dt',DateTime.strptime(md,'%d/%b/%Y:%H:%M:%S').strftime('%Y%m%d'))

"

}

}

output {

datahub {

access_id => "输入您的access_id"

access_key => "输入您的access_key"

endpoint => "需要根据自己的网络情况输入对应的endpoint"

project_name => "输入您的DataHub Project名称"

topic_name => "输入您对应的DataHub Topic"

#shard_id => "0"

#shard_keys => ["thread_id"]

dirty_data_continue => true

dirty_data_file => "/Users/yangyi/logstash-2.4.1/sample/dirty.data"

dirty_data_file_max_size => 1000

}

}