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

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具体路径下。

gzip -d GeoLiteCity.dat.gz
mv GeoLiteCity.dat /etc/logstash/.

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

input {

    file {

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

        start_position => "beginning"

    }

}

 

filter{

    grok {

        match => {

           "message" => "(?<ip>[^ ]*) - (?<user>[- ]*) \[(?<accesstime>[^\])*]\] \"(?<method>\S+)(?: +(?<url>[^\"]*?)(?: +\S*)?)?(?: +(?<protocol>[^\"]*))\" (?<status>[^ ]*) (?<byte_cnt>[^ ]*) \"(?<referer>[^\"]*)\" \"(?<agent>[^\"]*)\""

        }

}

    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

    }

}

配置文件为coolshell_log.conf。具体DataHub Topic信息可详见 数据存储 章节。

步骤7      启动任务示例如下：

 bin/logstash -f sample/coolshell_log.conf

【数据表创建】

附RDS创建表DDL：

---创建adm_refer_info---
 CREATE TABLE IF NOT EXISTS adm_refer_info(referer VARCHAR(32) PRIMARY KEY, referer_count BIGINT);
 
--创建adm_user_measures--
 CREATE TABLE IF NOT EXISTS adm_user_measures(device VARCHAR(32) PRIMARY KEY, pv BIGINT,uv BIGINT);
 
--创建adm_region_measures --
CREATE TABLE `adm_region_measures` (
  `region` varchar(32) NOT NULL,
  `region_cnt` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`region`)
)
 
--创建adm_region_measures --
CREATE TABLE `adm_region_measures` (
  `region` varchar(32) NOT NULL,
  `region_cnt` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`region`)
)
 
--创建flyingline_coordinates --
CREATE TABLE `flyingline_coordinates` (
  `city_name` varchar(32) DEFAULT NULL,
  `coordinates` varchar(50) DEFAULT NULL,
  `x` double DEFAULT NULL,
  `y` double DEFAULT NULL
)

【流式数据处理】

注册数据存储包括DataHub和RDS：

按照数据链路图中来编写处理逻辑（附核心代码）：

【处理逻辑1】

INSERT INTO coolshell_log_detail SELECT
    ip,
    accesstime,
    method,
    url,
    protocol,
    status,
    byte_cnt,
    regexp_extract(referer, '^[^/]+://([^/]+){1}') as referer,
    agent,
    CASE
            WHEN TOLOWER(agent) RLIKE 'android' THEN 'android'
            WHEN TOLOWER(agent) RLIKE 'iphone' THEN 'iphone'
            WHEN TOLOWER(agent) RLIKE 'ipad' THEN 'ipad'
            WHEN TOLOWER(agent) RLIKE 'macintosh' THEN 'macintosh'
            WHEN TOLOWER(agent) RLIKE 'windows phone' THEN 'windows_phone'
            WHEN TOLOWER(agent) RLIKE 'windows' THEN 'windows_pc'
            ELSE 'unknown'
     END AS device,
     CASE
            WHEN TOLOWER(agent) RLIKE '(bot|spider|crawler|slurp)' THEN 'crawler'
            WHEN TOLOWER(agent) RLIKE 'feed'
            OR url RLIKE 'feed' THEN 'feed'
            WHEN TOLOWER(agent) NOT RLIKE '(bot|spider|crawler|feed|slurp)'
            AND agent RLIKE '^[Mozilla|Opera]'
            AND url NOT RLIKE 'feed' THEN 'user'
            ELSE 'unknown'
     END AS identity
     FROM coolshell_log_tracker
       WHERE url NOT LIKE '^[/]+wp-';   

【处理逻辑2】

INSERT INTO coolshell_log_fact select
    md5(concat(ip, device, protocol, identity, agent)),--根据ip、device、protocol、identity和agent字段可以唯一确定uid
    ip,
    accesstime,
    method,
    url,
    protocol,
    status,
    byte_cnt,
    referer,
    agent,
    device,
    identity
    FROM coolshell_log_detail;

---adm_refer_info中的处理逻辑---
REPLACE INTO adm_refer_info SELECT
    referer,
COUNT(referer) as referer_count
FROM coolshell_log_fact
WHERE LENGTHqi(referer) > 1
GROUP BY referer;

--adm_user_measures中的处理逻辑---
REPLACE INTO adm_user_measures SELECT
    device,
    COUNT(uid) as pv,
    COUNT(distinct uid) as uv
    FROM coolshell_log_fact
    GROUP BY device;


附录：adm_region_measures和flyingline_coordinates处理逻辑
REPLACE INTO adm_region_measures SELECT
    CASE 
		WHEN region='%{[geoip][city_name]}' THEN 'unknown'
		WHEN region!='%{[geoip][city_name]}' THEN region
		END AS region,
count(region) FROM coolshell_log_tracker_bak
GROUP BY region;


INSERT INTO flyingline_coordinates
SELECT CASE 
		WHEN region='%{[geoip][city_name]}' THEN 'unknown'
		WHEN region!='%{[geoip][city_name]}' THEN region
		END AS region,
		coordinates,x,y FROM coolshell_log_tracker_bak where coordinates is NOT NULL;