大数据workshop：《在线用户行为分析：基于流式计算的数据处理及应用》之《流数据处理：通过StreamSQL分析用户行为》篇

2022-04-17 11:35:05

大数据workshop：《在线用户行为分析：基于流式计算的数据处理及应用》之《流数据处理：通过StreamSQL分析用户行为》篇

实验背景介绍

了解更多2017云栖大会·成都峰会 TechInsight & Workshop.

本手册为云栖大会Workshop之《在线用户行为分析：基于流式计算的数据处理及应用》场的《流数据处理：通过StreamSQL分析用户行为》篇所需。主要帮助现场学员熟悉并掌握阿里云流计算StreamCompute的操作和使用。

实验涉及大数据产品

阿里云流计算StreamCompute

前提准备

必备条件：如果您已经创建了StreamCompute Project，即可跳过创建StreamCompute项目空间章节，直接进入通过StreamSQL分析用户行为章节。

创建StreamCompute项目空间

目前阿里云流计算StreamCompute处于邀测阶段，为让大家深度体验到StreamCompute的便利性，在实验开始之前，幕后码农已为大家开通。

step1：保证账号在登录状态，点击进入阿里云流计算首页。
step2：点击创建项目，跳出创建项目对话框。
step3：填写项目名称、项目备注，点击创建。创建项目空间成果后，系统会自动刷新进入如下页面。
step4：创建项目空间成果后，系统会自动刷新进入如下页面，并点击开始使用进入流计算。

项目名需要字母或下划线开头，只能包含字母下划线和数字。

【注意】建议项目名称与DataHub Project规则保持一致为workshop_abc,其中abc为账号后三位数。

过StreamSQL分析用户行为

流计算角授权

step1：进入阿里云流计算开发平台，点击前往授权。

step2：在云资源访问授权页面，点击同意授权。

编写StreamSQL分析数据

step1：点击顶部菜单栏中的开发，进入开发页面。

step2：右键选择新建文件夹，在弹出框中填写文件夹名称为weblog（可根据需求命名），点击创建。

step3：在新建的weblog文件夹上右键选择新建作业，在新建作业弹出框中命名作业名称为workshop（可根据需求命名），点击新建。

step4：修改代码备注信息，进入编写StreamSQL开始分析数据。

step5：将DataHub Topic作为“输入表引用”，具体说明如下：

为了大家能够体验流式计算的实时计算能力，在下一章节大屏中也能具象体会到数据的实时刷新，本workshop特意为大家准备了一个实时产生日志的DataHub Topic。大家直接引用即可。

CREATE STREAM TABLE ods_log (
    remote                 STRING,
    ident                  STRING,
    auth                   STRING,
    dt                     STRING,
    method                 STRING,
    requestpath            STRING,
    httpversion            STRING,
    code                   STRING,
    bytes                  STRING,
    referrer               STRING,
    agent                  STRING,
    city                   STRING,
    province               STRING
) WITH (
        type='datahub',
        endpoint='http://dh-cn-hangzhou-internal.aliyuncs.com',
        accessId='LTAICO76WC1rREmW',
        accessKey='a89c1T96KMA2tOrEnb0SiBOA7GMMP3',
        projectName='workshop',
        topic='nginx_log'
);

step6：需要对源头DataHub的数据进行清洗和加工。逻辑如下：

创建临时表

CREATE TMP TABLE adm_log (
    remote                 STRING,
    identity               STRING,
    device                 STRING,
    dt                     STRING,
    method                 STRING,
    url                    STRING,
    httpversion            STRING,
    code                   STRING,
    bytes                  STRING,
    referrer               STRING,
    agent                  STRING,
    city                   STRING,
    province               STRING
);

创建处理逻辑

INSERT INTO adm_log SELECT
    remote,
    CASE 
        WHEN TOLOWER(agent) RLIKE '(bot|spider|crawler|slurp)' THEN 'crawler'
        WHEN TOLOWER(agent) RLIKE 'feed'
        OR requestpath RLIKE 'feed' THEN 'feed'
        WHEN TOLOWER(agent) NOT RLIKE '(bot|spider|crawler|feed|slurp)'
        AND agent RLIKE '^\"[Mozilla|Opera]'
        AND requestpath NOT RLIKE 'feed' THEN 'user'
        ELSE 'unknown'
    END AS identity,
    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,
    dt,
    method,
    requestpath,
    httpversion,
    code,
    bytes,
    regexp_extract(referrer, '^[^/]+://([^/]+){1}') as referrer,
    agent,
    city,
    province
    FROM ods_log
    WHERE requestpath NOT LIKE '^[/]+wp-';  --wp来自于主体请求的其他附加请求

step7：统计每天的实时浏览量pv的计算逻辑分别如下。

统计每小时的实时浏览量pv、统计各设备实时访问量、统计请求来源分布、统计热力图分布的计算逻辑此次操作中将不涉及，讲师会深度解析SQL处理逻辑，待大会结束后会统一将源码在此贴中披露供大家学习。请认证听讲师分享他的使用之道哦~~

每一个计算逻辑都需要“输出表引用”，即引用上一章节《实时数据分析：海量日志数据多维透视》创建的AnalyticDB表。特别注意需要替换自建创建的输出表名。

统计每天的实时浏览量pv

SQL逻辑如下：

CREATE RESULT TABLE pv_day (
    pv_time             STRING,
    pv_count            BIGINT,
    PRIMARY KEY (pv_time)
) WITH (
    type='ads',
    url='jdbc:mysql://workshop-demo-114ae71f.cn-hangzhou-1.ads.aliyuncs.com:10078/workshop_demo',
    username='LTAICO76WC1rREmW',
    password='a89c1T96KMA2tOrEnb0SiBOA7GMMP3',
    tableName='pv_day_620'  --记得替换成自己的AnalyticDB表名
);

REPLACE INTO pv_day
SELECT
    SUBSTR(dt, 1, 8) as pv_time,
    COUNT(1) as pv_count
FROM
    adm_log
GROUP BY
    SUBSTR(dt, 1, 8);