实训笔记8.30

8.30笔记
- 一、项目开发流程一共分为七个阶段
- - 1.1 数据产生阶段
  - 1.2 数据采集存储阶段
  - 1.3 数据清洗预处理阶段
  - 1.4 数据统计分析阶段
  - 1.5 数据迁移导出阶段
  - 1.6 数据可视化阶段
- 二、项目数据清洗预处理的实现
- - 2.1 清洗预处理规则
  - 2.2 代码实现
- 三、项目的数据统计分析阶段
- - 3.1 概念和技术选项
  - 3.2 Hive数据仓库进行统计分析时两个核心概念
  - - 3.1.1 数据仓库分层
    - 3.2.2 数据仓库建模
  - 3.3 数据统计分析的实现（最好把所有的HQL代码写到一个SQL文件中，最后统一执行运行）统计分析必须启动HDFS和YARN
  - - 3.3.1 构建ODS层
    - 3.3.2 构建DWD层
    - 3.3.3 构建ADS层
- 四、模拟其他年份，其他月份，其他日期的数据
- 五、相关代码

8.30笔记

一、项目开发流程一共分为七个阶段

1.1 数据产生阶段

1.2 数据采集存储阶段

1.3 数据清洗预处理阶段

1.4 数据统计分析阶段

1.5 数据迁移导出阶段

1.2~1.5：大数据开发阶段

1.3~1.5：周期性调度执行，三个阶段需要通过azkaban任务调度工具进行自动化周期调度执行，
（项目的第7个阶段，任务调度阶段）

1.6 数据可视化阶段

二、项目数据清洗预处理的实现

2.1 清洗预处理规则

数据清洗规则

一条用户行为数据如果字段个数不足16，那么数据不完整，舍弃
一条用户行为数据中如果响应状态码大于等于400的，那么数据访问错误，舍弃
一条用户行为数据中省份纬度经度年龄以-填充的，那么代表数据缺失，舍弃

数据预处理规则
预处理规则：清洗完成的数据中最后在输出时，有很多的字段我们不需要的，因此我们需要对部分数据进行舍弃，对需要保留的字段数据以\001特殊字符分割输出

技术选项：MapReduce技术

2.2 代码实现

【注意】

因为我们只需要做数据的清洗预处理操作，不涉及到聚合操作，因此我们只需要一个Mapper阶段即可，不需要reduce阶段
MapReduce数据清洗预处理是周期性调度执行的，一天执行一次，第二天处理前一天采集存储的数据，前一天采集存储的数据是以时间为基准的动态目录下存放，因此MR程序处理数据时，输入数据的目录必须得是昨天时间的目录。 【注意】会在第二天处理前一天的数据，一般会在第二天的凌晨去处理第一天采集存储的数据。（任务调度的事情）
MR程序处理完成的数据输出到HDFS上，但是数据清洗预处理完成的数据给Hive做统计分析的，Hive我们也是一天执行一次，Hive是在数据清洗预处理完成之后执行的。 MR程序处理完成的数据输出到HDFS上时，也必须以基于时间的动态目录存放

创建Maven项目，引入MR的编程依赖
编写MR程序的Mapper程序和Driver驱动程序
在本地测试运行无问题之后，需要将代码打成jar包上传到大数据环境中在YARN上运行必须启动YARN

三、项目的数据统计分析阶段

3.1 概念和技术选项

统计分析就是基于我们清洗预处理完成的高质量，从不同的数据纬度聚合数据，或者对数据进行计算得到我们感兴趣的一些指标或者是对网站运营发展有关的一些指标。

统计分析进行数据计算时，可能涉及到大量的聚合操作以及一些排名、排序等等操作，而这些操作也都是数据计算，那么我们就可以使用大数据计算框架完成，而大数据计算框架MapReduce如果要聚合、排序、分组等操作，MR代码就会非常的复杂。因此我们一般做统计分析时有一个想法，既能计算大量的数据，还能快速简单的进行数据的聚合、排名、分组等操作。就可以使用Hive数据仓库技术完成。

【注意】基本上到现在为止，如果我们要做大数据统计分析，不是直接使用大数据计算框架（MapReduce、Spark、Flink），因为大数据统计分析涉及到大量的聚合、排序、分组等等操作，操作如果直接使用大数据计算框架代码会非常的复杂。基本上都是使用类SQL（表面上写的是类SQL语句，底层还是大数据计算框架）的方式进行大数据统计分析的。

Hadoop—Hive

Spark—Spark SQL

Flink—Flink SQL

3.2 Hive数据仓库进行统计分析时两个核心概念

3.1.1 数据仓库分层

数据仓库建模是用来梳理表和表之间的关系的，便于我们后期进行统计分析。数据仓库分层是我们使用数据仓库进行统计分析的开发流程。

数据仓库分层从最底层开始到最高层主要有如下三层（不同的公司基于三层更加细致的分层）

ODS层（数据贴源层）

如果我们要使用Hive数据仓库做统计分析，首先我们需要把清洗预处理完成的数据导入到Hive中加载成为一个数据表，ODS层指的就是把清洗预处理完成的数据原模原样的导入到Hive中，导入进来之后这些表组成了ODS层

DW层（数据仓库层）–Hive统计分析的核心数据仓库建模的阶段
1. DWD层（明细宽表层）
  
  把ODS层的数据表可以再次处理一下构建成为一个明细宽表
  
  明细宽表一般会把ODS层的字段拆分成更加细粒度的字段，便于我们后期好做统计分析（时间字段）
2. DIM层（纬表层）

纬度表如果比较多，那么纬度表单独划分到DW的DIM层

ADS层（数据应用层）

将统计分析的结果以指标表的形式存储到ADS层

3.2.2 数据仓库建模

建模的目的是为了方便我们后期统计分析

在使用Hive进行数据统计分析时，首先必须先把清洗预处理完成的数据加载到Hive中成为数据表，而且一般在真实的企业项目中，清洗预处理完成数据不止一个，各种各样的数据，数据和数据之间都是有关系的。

所谓的数据仓库建模就是我们在对数据进行清洗预处理的时候，清洗预处理完成之后的多个数据之间的关系梳理建模

数据仓库建模的名词解释
1. 事实表：一张表中基本全都是外键，如果我们需要查询数据，需要将这个表和各个对应的其它数据表进行关联查询才能得到我们想要的数据订单表
2. 维度表：事实表中外键对应的详细信息存储的表，而且他也是我们统计分析时纬度信息用户表商品表
数据仓库模型建立有很多种方式的，主要分为
1. 3NF数据仓库建模
2. 纬度建模
  1. 星型模型
    
    事实表直接与纬度表关联，而且只有一级关联
  2. 雪花模型
    
    事实表直接与维度表关联，纬度表拆分出更加细致的一些纬度表
  3. 星座模型
    
    在一个数仓中，事实表有多个，每一个事实表都有它自己对应的纬度表，纬度表还有它的二级纬度表

如何完成建模？数据清洗预处理的时候，把数据处理成为合适的模型结构

3.3 数据统计分析的实现（最好把所有的HQL代码写到一个SQL文件中，最后统一执行运行）统计分析必须启动HDFS和YARN

3.3.1 构建ODS层

ODS层指的是我们把清洗预处理完成的数据不加以任何的处理，直接原模原样的在Hive中构建与之对应的表格，并且把数据装载到表格当中

清洗预处理完成的数据格式以\001特殊字符分割的，这样的话可以避免分隔符和字段的中一些符号冲突，导致装载数据到Hive出现串行的问题。

Hive中数据表有很多分类的：内部表、外部表、分区表、分桶表

考虑：数据统计分析一天执行一次，也就意味着我们每天处理完成的数据都需要往Hive的ODS层的数据表导入一份，如何区分ODS层导入的数据是哪一天？需要构建一个分区表（基于时间的）。外部表

3.3.2 构建DWD层

DWD明细宽表层就是把ODS层的数据表字段拆分成为更加细粒度的字段，便于我们后期的统计分析。 DWD层说白了就是在ODS的数据表基础之上在多增加一些冗余字段，但是方便我们后期操作了

ODS层的字段如下：

可以拆分的字段主要有两个
1. 时间字段：后期需要基于细粒度的时间做统计分析
  
  年
  
  月
  
  日
  
  时
2. 来源URL字段：后期统计站内站外的流量占比，站内站外的对比是基于HOST主机名/域名——HOST

DWD层这个数据表就属于我们Hive的自有表了，因此明细宽表我们构建成为内部分区表即可

明细宽表中没有数据，明细宽表中的数据从什么地方来？因为DWD层是基于ODS层建立的，因此DWD层的数据需要从ODS层查询获得。需要从ODS层对应的数据表中查询指定的数据添加到DWD层当中（注意一下分区的问题）。

3.3.3 构建ADS层

ADS层其实就是我们基于DW数据仓库库构建的DWD和DIM层的数据表，进行查询，通过聚合、分组、排序等等操作统计相关的指标，得到指标数据，然后将指标数据存储到一个Hive数据表中。

基于时间纬度的指标
1. 统计网站每年的用户的流量
  
  网站每天都会产生数据，每一天数据一增加，那么当前年份的用户访问量必然增加一天的数据
  
  思路：不是针对明细宽表某一个分区的数据进行统计分析，而是针对于明细宽表中整体数据集进行统计分析（所有的分区进行操作）
  
  实现：因为在明细宽表中已经拆分除了visit_year字段，因此我们只需要根据visit_year分区聚合数据即可得到，每一年的用户访问量
  
  select visit_year,count(*) from dwd_user_behavior_detail group by visit_year;
2. 统计网站每一年不同月份的用户流量
  1. 实现同上
  2. 区别：分组时，需要根据年和月来分组
3. 统计网站每一年不同月份下每天用户的访问量
  1. 实现同上
  2. 区别：分组的时候，需要根据年、月、日三个字段来分组
4. 统计网站每一年不同月份下每天的每小时用户的访问量
  1. 实现同上
  2. 区别：分组的时候，需要根据年、月、日、时四个字段来分组
5. 统计网站每一年每一个月的流量相比于上个月的比例：开窗函数（上边界和下边界），针对每一年不同月份的用户流量指标的二次分析结果
基于地理纬度的指标
1. 统计每天网站不同省份用户的访问量
2. 每天访问量TOP10的省份
基于用户纬度的指标
1. 统计不同年龄段用户的访问量
2. 统计每天网站的独立访客数
基于终端纬度的指标——统计网站用户使用的不同浏览器的占比情况
基于来源纬度的指标——统计网站每天站内和站外的流量占比
指标有很多，可以进行各种自由扩展

四、模拟其他年份，其他月份，其他日期的数据

只需要通过date -s “时间” 系统时间改成我们想要模拟数据的日期即可

把以前产生的userBehavior.log文件删除了

然后启动采集程序启动数据模拟程序

处理数据，只需要再把系统时间往后调整一天

五、相关代码

-- 0、创建一个项目专属的数据库
create database if not exists project;
use project;
-- 1、构建ODS层的数据表，数据表和清洗预处理完成的数据格式一致的表格 而且ODS层的表格是外部分区表
create external table if not exists ods_user_behavior_origin(
   ip_addr string, --ip地址
   visit_time string,-- 浏览时间
   request_url string,-- 行为触发之后的请求网址
   referer_url string,-- 来源网址
   user_agent string,-- 用户使用的浏览器信息
   province string, --省份
   latitude string, -- 纬度
   longitude string, -- 经度
   age int --年龄
)partitioned by(data_gen_time string)
row format delimited fields terminated by '\001'; 

-- 2、需要将清洗预处理完成的昨天的数据（/dataClean/yyyy-MM-dd）导入到ods层的昨天时间分区中。
load data inpath '/dataClean/${hiveconf:yesterday}' into table ods_user_behavior_origin partition(data_gen_time='${hiveconf:yesterday}')

-- 3、构建DWD明细宽表层  就是在ods数据表基础之上增加了五列字段 visit_year  visit_month  visit_day  visit_hour referer_host
create  table if not exists dwd_user_behavior_detail(
   ip_addr string, --ip地址
   visit_time string,-- 浏览时间
   visit_year string, --拆分的浏览年份
   visit_month string,----拆分的浏览月份
   visit_day string,---拆分的浏览天
   visit_hour string,---拆分的浏览时
   request_url string,-- 行为触发之后的请求网址
   referer_url string,-- 来源网址
   referer_host string, -- 来源网址的域名
   user_agent string,-- 用户使用的浏览器信息
   province string, --省份
   latitude string, -- 纬度
   longitude string, -- 经度
   age int --年龄
)partitioned by(data_gen_time string)
row format delimited fields terminated by '\001'; 

-- 4、从贴源数据表查询明细宽表所需的数据，然后把数据增加到明细宽表的昨天的时间分区中
insert overwrite table dwd_user_behavior_detail partition(data_gen_time='${hiveconf:yesterday}')
select 
	ip_addr,
	visit_time,
	date_format(visit_time,'yyyy') as visit_year,
	date_format(visit_time,'MM') as visit_month,
	date_format(visit_time,'dd') as visit_day,
	date_format(visit_time,'HH') as visit_hour,
	request_url,
	referer_url,
	parse_url(referer_url,'HOST') as referer_host,
	user_agent,
	province,
	latitude,
	longitude,
	age
from ods_user_behavior_origin 
where data_gen_time='${hiveconf:yesterday}';

-- 5、构建数据应用层 统计各种各样的指标数据，并且把指标数据保存到Hive对应的指标表中 指标表的结构必须和我们查询的指标数据一致的
--（1）基于时间纬度--统计网站每年用户的流量
create table if not exists ads_year_flow(
   visit_year string,
   flow bigint
)row format delimited fields terminated by '\001';

-- 将统计的结果覆盖添加到基于年份的指标表中 防止年份流量数据重复
insert overwrite table ads_year_flow
select  visit_year,count(*)  from dwd_user_behavior_detail group by visit_year;


-- （2）统计网站每一年不同月份的用户流量
create table if not exists ads_month_flow(
   visit_year string,
   visit_month string,
   flow bigint
)row format delimited fields terminated by '\001';

insert overwrite table ads_month_flow
select  visit_year,visit_month,count(*)  from dwd_user_behavior_detail group by visit_year,visit_month;

package com.sxuek;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * MR程序的驱动程序
 *   封装MR程序的
 */
public class DataCleanDriver {
    public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException, ClassNotFoundException {
        //1、准备一个配置文件对象 用来封装MR程序运行的一些相关配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.38.101:9000");

        //2、准备一个用于封装MR程序的Job对象
        Job job = Job.getInstance(conf);
        //3、设置一个jar包的位置判断 是为了让MR程序打成jar包以后可以正常运行
        job.setJarByClass(DataCleanDriver.class);

        //4、配置MR程序的InputFormat以及我们的输入文件路径
        job.setInputFormatClass(TextInputFormat.class);
        //输入文件路径是采集存储的昨天的数据目录 /dataCollect/yyyy-MM-dd 在驱动程序中获取昨天的时间
        Calendar cal = Calendar.getInstance();//默认是今天的时间
        cal.add(Calendar.DAY_OF_MONTH,-1);
        Date date = cal.getTime();
        //需要把获取的昨天的时间转换称为yyyy-MM-dd这种格式
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        String yesterday = sdf.format(date);
        String inputPath = "/dataCollect/"+yesterday;
        System.out.println("待处理的数据目录为"+inputPath);
        TextInputFormat.setInputPaths(job,new Path(inputPath));

        //5、封装Mapper阶段
        job.setMapperClass(DataCleanMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //6、封装Reducer阶段
        job.setNumReduceTasks(0);

        //7、封装OutputFormat  封装输出目录 输出目录也得是一个动态目录 /dataClean/yyyy-MM-dd
        job.setOutputFormatClass(TextOutputFormat.class);
        Path outPath = new Path("/dataClean/"+yesterday);
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.38.101:9000"), conf, "root");
        if (fileSystem.exists(outPath)){
            fileSystem.delete(outPath,true);
        }
        TextOutputFormat.setOutputPath(job,outPath);

        //8、提交MR程序运行 然后运行成功合理的关闭MR程序
        boolean flag = job.waitForCompletion(true);
        System.exit(flag?0:1);
    }
}

package com.sxuek;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * MR程序的Mapper阶段
 *   四个泛型：Mapper阶段的输入key value  输出key value
 *   因为采集存储的数据是以纯文本文档的形式在HDFS上存放的，而且一个文件采集存储的时候设置128M
 *   因此我们清洗预处理时，一个文件正好可以当作一个切片使用，不存储大量的小文件
 *   所以我们MR程序的InputFormat我们直接使用默认TextInputFormat机制即可
 *
 * 因为数据清洗预处理只需要mapper阶段，不需要reduce阶段，因此mapper阶段的输出就是最终输出
 * mapper输出的就是每一行清洗预处理完成的一行数据
 */
public class DataCleanMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        /**
         * 1、获取到切片中对应的每一行的用户行为日志数据
         * 172.83.232.165 - - 2023-08-29 15:19:32 "POST https://www.bailongma.com/category/b HTTP/1.1" 200 15075 https://www.baidu.com/search Mozilla/5.0 (Macintosh; PPC Mac OS X; U; en) Opera 8.0 广东 23.08 113.14 56
         */
        String line = value.toString();
        /**
         * 2、将用户行为数据以空格分割得到一个数组 数组中存放的就是用户行为数据对应的多个字段
         */
        String[] array = line.split(" ");
        //拿到数组之后 我们应该先进行一个字段个数不合法的清洗过程
        if (array.length < 16){
            return;
        }

        //获取用户行为数据中各种字段信息
        //获取用户的访问IP 第一个字段
        String ip = array[0];
        //获取用户的访问时间  第四个字段和第五字段两部分组成
        String time = array[3]+" "+array[4];
        //获取用户的请求URL  第七个字段
        String requestURL = array[6];
        //获取用户的响应状态码  第九个字段 而且是整数类型
        int code = Integer.parseInt(array[8]);
        //获取用户的来源URL  第11个字段
        String refererURL = array[10];
        //获取用户的浏览器信息 字段个数不确定  但是浏览器的位置一定是从第12个字段开始到倒数第五个字段结束
        StringBuffer stringBuffer = new StringBuffer();
        for (int i = 11; i <= array.length-5 ; i++) {
            stringBuffer.append(array[i]);
        }
        String userAgent = stringBuffer.toString();
        //获取用户的省份 纬度 经度 年龄
        String province = array[array.length - 4];
        String latitude = array[array.length - 3];
        String longitude = array[array.length - 2];
        String age = array[array.length - 1];

        /**
         *  3、清洗过程
         */
        //清洗状态码大于等于400的
        if (code >= 400){
            return;
        }
        //把地理位置信息和用户年龄为-的数据舍弃
        if (province.equals("-") || latitude.equals("-") || longitude.equals("-") || age.equals("-")){
            return;
        }

        /**
         *  4、将数据预处理成我们想要的格式
         *  ip,访问时间,请求URL,来源URL,用户浏览器,省份,纬度,经度,年龄
         */

        StringBuffer sb = new StringBuffer();
        //append可以一直拼接  链式编程 方法执行完成返回当前方法的调用对象
        sb.append(ip).append(",").append(time).append(",").append(requestURL).append(",")
                .append(refererURL).append(",").append(userAgent).append(",").append(province)
                .append(",").append(latitude).append(",").append(longitude).append(",").append(age);
        String result = sb.toString();
        //5、输出预处理之后的结果数据即可
        context.write(new Text(result),NullWritable.get());
    }
}