数据仓库是为企业决策提供数据支撑的,进行存储、分析与计算。
数据仓库的数据来源有:业务数据、用户行为数据、爬虫数据等。
(1)项目需求
1、用户行为数据采集平台搭建
2、业务数据采集平台搭建
3、数据仓库维度建模
4、分析,设备、会员、商品、地区、活动等电商核心主题,统计的报表指标近100个。
5、采用即席查询工具,随时进行指标分析
6、对集群性能进行监控,发生异常需要报警。【监控每一个框架的进程的好坏】
7、元数据管理
8、质量监控【监控的是数据的异常变化,例如能够实时监控今天的指标和昨天的指标的差异变化】
9、权限管理
需要考虑的问题:
1、项目技术如何选型?
2、框架版本如何选型(Apache、CDH、HDP)
3、服务器使用物理机还是云主机?
4、如何确定集群规模?(假设每台服务器8T硬盘)
技术选型主要考虑因素:数据量大小、业务需求、行业内经验、技术成熟度、开发维护成本、总成本预算
数仓流程架构图:
框架发行版本选择:
如何选择Apache/CDHHDP版本?【免费】
(1) Apache:运维麻烦,组件间兼容性需要自己调研。(一般大厂使用,技术实力雄厚,有专业的运维人员)(建议使用)
(2)CDH:国内使用最多的版本,但CM不开源,今年开始收费,一个节点1万美金/年。
(3)HDP:开源,可以进行二次开发,但是没有CDH稳定,国内使用较少
云服务选择【收费】
(1)阿里云的EMR、MaxCompute、DataWorks
(2)亚马逊云EMR
(3)腾讯云ENMR
(4)华为云EMR
服务器选择:
集群资源规划设计:
在企业中通常会搭建一套生产集群和一套测试集群。生产集群运行生产任务,测试集群用于上线前代码编写和测试。
我们要收集和分析的数据主要包括页面数据、事件数据、曝光数据、启动数据和错误数据。
1、页面浏览记录
页面浏览记录,记录的是访客对页面的浏览行为,该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及页面信息等。
2、动作记录
动作记录,记录的是用户的业务操作行为,该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息 及动作目标对象信息等。
3、曝光记录
曝光记录,记录的是曝光行为,该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及曝光对象信息等。
4、启动记录
启动记录,记录的是用户启动应用的行为,该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、启动类型及开屏广告信息等。
5、错误记录
启动记录,记录的是用户在使用应用过程中的报错行为,该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、以及可能与报错相关的页面信息、动作信息、曝光信息和动作信息。
用户行为日志的内容,主要包括用户的各项行为信息以及行为所处的环境信息。收集这些信息的主要目的是优化产品和为各项分析统计指标提供数据支撑。收集这些信息的手段通常为埋点。
目前主流的埋点方式,有代码埋点(前端/后端)、可视化埋点、全埋点等。
代码埋点是通过调用埋点SDK函数,在需要埋点的业务逻辑功能位置调用接口,上报埋点数据。例如,我们对页面中的某个按钮埋点后,当这个按钮被点击时,可以在这个按钮对应的 OnClick 函数里面调用SDK提供的数据发送接口,来发送数据。
可视化埋点只需要研发人员集成采集 SDK,不需要写埋点代码,业务人员就可以通过访问分析平台的“圈选”功能,来“圈”出需要对用户行为进行捕捉的控件,并对该事件进行命名。圈选完毕后,这些配置会同步到各个用户的终端上,由采集 SDK 按照圈选的配置自动进行用户行为数据的采集和发送。
全埋点是通过在产品中嵌入SDK,前端自动采集页面上的全部用户行为事件,上报埋点数据,相当于做了一个统一的埋点。然后再通过界面配置哪些数据需要在系统里面进行分析。
埋点数据上报时机:本次项目采用方式一埋点。
埋点数据日志结构:
我们的日志结构大致可分为两类,一是普通页面埋点日志,二是启动日志。
普通页面埋点日志格式如下:
{
"common": { -- 环境信息
"ar": "230000", -- 地区编码
"ba": "iPhone", -- 手机品牌
"ch": "Appstore", -- 渠道
"is_new": "1", -- 是否首日使用,首次使用的当日,该字段值为1,过了24:00,该字段置为0。
"md": "iPhone 8", -- 手机型号
"mid": "YXfhjAYH6As2z9Iq", -- 设备id
"os": "iOS 13.2.9", -- 操作系统
"uid": "485", -- 会员id
"vc": "v2.1.134" -- app版本号
},
"actions": [{ -- 动作(事件)
"action_id": "favor_add", -- 动作id
"item": "3", -- 目标id
"item_type": "sku_id", -- 目标类型
"ts": 1585744376605 -- 动作时间戳
}
],
"displays": [{ -- 曝光
"displayType": "query", -- 曝光类型
"item": "3", -- 曝光对象id
"item_type": "sku_id", -- 曝光对象类型
"order": 1, -- 出现顺序
"pos_id": 2 -- 曝光位置
},
{
"displayType": "promotion",
"item": "6",
"item_type": "sku_id",
"order": 2,
"pos_id": 1
},
{
"displayType": "promotion",
"item": "9",
"item_type": "sku_id",
"order": 3,
"pos_id": 3
},
{
"displayType": "recommend",
"item": "6",
"item_type": "sku_id",
"order": 4,
"pos_id": 2
},
{
"displayType": "query ",
"item": "6",
"item_type": "sku_id",
"order": 5,
"pos_id": 1
}
],
"page": { -- 页面信息
"during_time": 7648, -- 持续时间毫秒
"item": "3", -- 目标id
"item_type": "sku_id", -- 目标类型
"last_page_id": "login", -- 上页类型
"page_id": "good_detail", -- 页面ID
"sourceType": "promotion" -- 来源类型
},
"err": { --错误
"error_code": "1234", --错误码
"msg": "***********" --错误信息
},
"ts": 1585744374423 --跳入时间戳
}
启动页面埋点日志格式如下:
{
"common": {
"ar": "370000",
"ba": "Honor",
"ch": "wandoujia",
"is_new": "1",
"md": "Honor 20s",
"mid": "eQF5boERMJFOujcp",
"os": "Android 11.0",
"uid": "76",
"vc": "v2.1.134"
},
"start": {
"entry": "icon", --icon手机图标 notice 通知 install 安装后启动
"loading_time": 18803, --启动加载时间
"open_ad_id": 7, --广告页ID
"open_ad_ms": 3449, -- 广告总共播放时间
"open_ad_skip_ms": 1989 -- 用户跳过广告时点
},
"err":{ --错误
"error_code": "1234", --错误码
"msg": "***********" --错误信息
},
"ts": 1585744304000
}
1)将application.yml、gmall2020-mock-log-2021-10-10.jar、path.json、logback.xml
上传到hadoop102的/home/wenxin/module/applog目录下
(1)创建applog路径
[root@hadoop102 module]$ mkdir /home/wenxin/module/applog
(2)上传文件到/opt/module/applog目录
3)生成日志
(1)进入到/opt/module/applog路径,执行以下命令
[root@hadoop102 applog]$ java -jar gmall2020-mock-log-2021-10-10.jar
(2)在/opt/module/applog/log目录下查看生成日志
[root@hadoop102 log]$ ll
在hadoop102的/home/atguigu目录下创建bin目录,这样脚本可以在服务器的任何目录执行。
[root@hadoop102 ~]$ echo $PATH
/usr/local/bin:/usr/bin:/usr/local/sbin:/usr/sbin:/home/atguigu/.local/bin:/home/atguigu/bin
(1)在/home/atguigu/bin目录下创建脚本lg.sh
[root@hadoop102 bin]$ vim lg.sh
(2)在脚本中编写如下内容
#!/bin/bash
for i in hadoop102 hadoop103; do
echo "========== $i =========="
ssh $i "cd /home/wenxin/module/applog/; java -jar gmall2020-mock-log-2021-10-10.jar >/dev/null 2>&1 &"
done
1)在/home/atguigu/bin目录下创建脚本xcall
[root@hadoop102 bin]$ vim xcall
2)在脚本中编写如下内容
#! /bin/bash
for i in hadoop102 hadoop103 hadoop104
do
echo --------- $i ----------
ssh $i "$*"
done
3)修改脚本执行权限
[root@hadoop102 bin]$ chmod 777 xcall
4)启动脚本
[root@hadoop102 bin]$ xcall.sh jps
①生产环境服务器磁盘情况
②在hdfs-site.xml文件中配置多目录,注意新挂载磁盘的访问权限问题。
HDFS的DataNode节点保存数据的路径由dfs.datanode.data.dir参数决定,其默认值为file://${hadoop.tmp.dir}/dfs/data,若服务器有多个磁盘,必须对该参数进行修改。如服务器磁盘如上图所示,则该参数应修改为如下的值。
<property>
<name>dfs.datanode.data.dir</name>
<value>file:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4</value>
</property>
注意:每台服务器挂载的磁盘不一样,所以每个节点的多目录配置可以不一致。单独配置即可。因为有可能hadoop102有4块硬盘,但是hadoop103只有两块硬盘。
①节点间数据均衡
开启数据均衡命令:
start-balancer.sh -threshold 10
对于参数10,代表的是集群中各个节点的磁盘空间利用率相差不超过10%,可根据实际情况进行调整。
停止数据均衡命令:
stop-balancer.sh
注意:HDFS需要启动单独的Rebalance Server来执行Rebalance操作,所以尽量不要在NameNode上执行start-balancer.sh,而是找一台比较空闲的机器。
②磁盘间数据均衡
生成均衡计划 但是其实虚拟机是只有一块硬盘的,不会生成计划。
hdfs diskbalancer -plan hadoop103
执行均衡计划
hdfs diskbalancer -execute hadoop103.plan.json
查看当前均衡任务的执行情况
hdfs diskbalancer -query hadoop103
取消均衡任务
hdfs diskbalancer -cancel hadoop103.plan.json
这部分的详情请参考这篇博客:Hadoop支持LZO压缩配置详细(附文件)
接下来测试LZO压缩是否好用:
创建一个路径:
[root@hadoop102 hadoop-3.1.3]# hadoop fs -mkdir /input
上传一个文件夹到当前目录:
[root@hadoop102 hadoop-3.1.3]# hadoop fs -put README.txt /input
执行wordCount:
[root@hadoop102 hadoop-3.1.3]# hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar wordcount -Dmapreduce.output.fileoutputformat.compress=true -Dmapreduce.output.fileoutputformat.compress.codec-com.hadoop.compression.lzo.LzopCodec /input /output
(1)将bigtable.lzo(200M)上传到集群的根目录
[root@hadoop102 module]$ hadoop fs -mkdir /input
[root@hadoop102 module]$ hadoop fs -put bigtable.lzo /input
(2)执行wordcount程序
[root@hadoop102 hadoop-3.1.3]# hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar wordcount -Dmapreduce.job.inputformat.class=com.hadoop.mapreduce.LzoTextInputFormat /input /output1
(3)对上传的LZO文件创建索引
[root@hadoop102 hadoop-3.1.3]# hadoop jar /home/wenxin/module/hadoop-3.1.3/share/
(4)再次执行wordcount程序
[root@hadoop102 hadoop-3.1.3]# hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar wordcount -Dmapreduce.job.inputformat.class=com.hadoop.mapreduce.LzoTextInputFormat /input /output1
(1)HDFS参数调优hdfs-site.xml
The number of Namenode RPC server threads that listen to requests from clients. If dfs.namenode.servicerpc-address is not configured then Namenode RPC server threads listen to requests from all nodes.
NameNode有一个工作线程池,用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。
对于大集群或者有大量客户端的集群来说,通常需要增大参数dfs.namenode.handler.count的默认值10。
<property>
<name>dfs.namenode.handler.count</name>
<value>10</value>
</property>
[root@hadoop102 ~]$ python
Python 2.7.5 (default, Apr 11 2018, 07:36:10)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import math
>>> print int(20*math.log(8))
41
>>> quit()
(2)YARN参数调优yarn-site.xml【京东面试题】
情景描述:总共7台机器,每天几亿条数据,数据源->Flume->Kafka->HDFS->Hive
面临问题:数据统计主要用HiveSQL,没有数据倾斜,小文件已经做了合并处理,开启的JVM重用,而且IO没有阻塞,内存用了不到50%。但是还是跑的非常慢,而且数据量洪峰过来时,整个集群都会宕掉。基于这种情况有没有优化方案。这是一个典型的集群资源没有利用到位,
解决办法:
yarn. nodemanager.resource.memory-mb NodeManager 使用内存数
yarn.nodemanager.resource.cpu-vcores NodeManager 使用CPU核数
FileChannel存储在磁盘当中,可靠性较高,但效率较低。
MemoryChannel存储在内存当中,可靠性较低,但效率较高。
在没有KafkaChannel的情况下:
1、如果是金融类公司、对钱要求非常准确的公司通常会选择FileChannel,保证数据的可靠性。
2、如果传输的是普通的日志信息,则通常选择MemoryChannel,比如一些公司每天丢100-200万条是很正常的。
#定义组件
a1.sources=r1
a1.channels=c1
a1.sinks=k1
#配置source
a1.sources.r1.type=org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics=topic_log
a1.sources.r1.batchSize=5000
a1.sources.r1.batchDurationMillis=2000
#时间戳拦截器
a1.sources.r1.interceptors=i1
a1.sources.r1.interceptors.i1.type=com.wenxin.gmall.flume.interceptor.TimeStampIntercepter$Builder
#配置channel
a1.channels.c1.type=file
a1.channels.c1.checkpointDir=/home/wenxin/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs=/home/wenxin/module/flume/data/behavior1/
#配置sink
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/origin_data/gmall/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix=log-
a1.sinks.k1.hdfs.round=false
#控制输出文件是原生文件
a1.sinks.k1.hdfs.fileType=CompressedStream
a1.sinks.k1khdfs.codeC=lzop
#拼接组件
a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1
package com.wenxin.gmall.flume.interceptor;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
public class ETLInterceptor implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
//1、获取body当中的数据并转成字符串
byte[] body = event.getBody();
String log = new String(body, StandardCharsets.UTF_8);
//2、判断字符串是否是一个合法的json,是:返回当前event;不是:返回null
if (JSONUtil.isJSONValidate(log)) {
return event;
} else {
return null;
}
}
@Override
public List<Event> intercept(List<Event> list) {
Iterator<Event> iterator = list.iterator();
while (iterator.hasNext()){
Event next = iterator.next();
if(intercept(next)==null){
iterator.remove();
}
}
return list;
}
public static class Builder implements Interceptor.Builder{
@Override
public Interceptor build() {
return new ETLInterceptor();
}
@Override
public void configure(Context context) {
}
}
@Override
public void close() {
}
}
package com.wenxin.gmall.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONException;
public class JSONUtil {
/*
* 通过异常判断是否是json字符串
* 是:返回true 不是:返回false
* */
public static boolean isJSONValidate(String log){
try {
JSONObject.parseObject(log);
return true;
}catch (JSONException e){
return false;
}
}
}
package com.wenxin.gmall.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
/**
* @author Susie-Wen
* @version 1.0
* @description:
* @date 2023/12/22 10:27
*/
public class TimeStampIntercepter implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
//将日志拦下来,取出header里面的key,取出body里面的对应的日志时间,将ts的值赋值给header的key timestamp
//1 获取header头
Map<String,String> headers=event.getHeaders();
//2 获取body中的ts
byte[] body=event.getBody();
String log =new String(body, StandardCharsets.UTF_8);
JSONObject jsonObject = JSONObject.parseObject(log);
String ts = jsonObject.getString("ts");
//3 将ts赋值给timestamp
headers.put("timestamp",ts);
return null;
}
@Override
public List<Event> intercept(List<Event> list) {
for(Event event:list){
intercept(event);
}
return list;
}
@Override
public void close() {
}
public static class Builder implements Interceptor.Builder{
@Override
public Interceptor build() {
return new TimeStampIntercepter();
}
@Override
public void configure(Context context) {
}
}
}
打包之后将其上传到hadoop104上
这个文件即包含之前写的ETL拦截器,也包含时间戳拦截器
#! /bin/bash
case $1 in
"start"){
for i in hadoop104
do
echo "----- 启动 $i 消费flume-----"
ssh $i "nohup /home/wenxin/module/flume/bin/flume-ng agent --conf-file /home/wenxin/module/flume/conf/kafka-flume-
hdfs.conf --name a1 -Dflume.root.logger=INFO,LOGFILE >/home/wenxin/module/flume/log2.txt 2>&1 &"
done
};;
"stop"){
for i in hadoop104
do
echo "-----停止$i消费flume-----"
ssh $i "ps -ef | grep kafka-flume-hdfs | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
done
};;
esac
#!/bin/bash
case $1 in
"start")
echo "----- 启动集群 -----"
zk.sh start
hdp.sh start
kf.sh start
f1.sh start
f2.sh start
;;
"stop")
echo "----- 停止集群 -----"
f2.sh stop
f1.sh stop
kf.sh stop
hdp.sh stop
zk.sh stop
;;
esac
[root@hadoop102 ~]$ cd $HADOOP_HOME/etc/hadoop
sbin/start-dfs.sh
sbin/start-yarn.sh
查看使用jps
hive --service metastore &
hive --service hiveserver2 &
beeline
!connect jdbc:hive2://hadoop102:10000
登录mysql:
[root@hadoop102 module]# mysql -uroot -p123456
启动zookeeper:
zookeeper在这个目录下:/home/wenxin/module
启动zookeeper:zk.sh start
停止zookeeper:zk.sh stop
查看zookeeper状态:zk.sh status 【必须要有leader和follower】
查看3台服务器的集群状态:xcall.sh jps
启动kafka:kf.sh start
停止kafka:kf.sh stop
启动flume:f1.sh start
停止flume:f1.sh stop
启动消费flume:f2.sh start
停止消费flume:f2.sh stop
生成日志:lg.sh
启动整个采集集群:cluster start
停止整个采集集群:clustar stop