网站建设运动会成绩管理系统,一级消防工程师考试报名,如何咨询网络服务商,网站运营公司第4章 用户行为数据采集模块
4.1 数据通道 4.2 环境准备
4.2.1 集群所有进程查看脚本
1#xff09;在/home/atguigu/bin目录下创建脚本xcall
[atguiguhadoop102 bin]$ vim xcall2#xff09;在脚本中编写如下内容
#! /bin/bashfor i in hadoop102 hadoop103 hadoop104
d…第4章 用户行为数据采集模块
4.1 数据通道 4.2 环境准备
4.2.1 集群所有进程查看脚本
1在/home/atguigu/bin目录下创建脚本xcall
[atguiguhadoop102 bin]$ vim xcall2在脚本中编写如下内容
#! /bin/bashfor i in hadoop102 hadoop103 hadoop104
doecho --------- $i ----------ssh $i $*
done3修改脚本执行权限
[atguiguhadoop102 bin]$ chmod 777 xcall4启动脚本
[atguiguhadoop102 bin]$ xcall jps4.2.2 Hadoop安装
1安装步骤
详见尚硅谷大数据技术之Hadoop入门
1集群规划
hadoop102hadoop103hadoop104HDFSNameNodeDataNodeDataNodeDataNodeSecondaryNameNodeYARNNodeManagerResourcemanagerNodeManagerNodeManager
注意尽量使用离线方式安装
2项目经验
1项目经验之HDFS存储多目录
①生产环境服务器磁盘情况 ②在hdfs-site.xml文件中配置多目录注意新挂载磁盘的访问权限问题
HDFS的DataNode节点保存数据的路径由dfs.datanode.data.dir参数决定其默认值为file://${hadoop.tmp.dir}/dfs/data若服务器有多个磁盘必须对该参数进行修改。如服务器磁盘如上图所示则该参数应修改为如下的值。
propertynamedfs.datanode.data.dir/namevaluefile:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4/value
/property注意因为每台服务器节点的磁盘情况不同所以这个配置配完之后不需要分发
2项目经验之集群数据均衡
①节点间数据均衡
开启数据均衡命令
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.json3项目经验之Hadoop参数调优
①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。
propertynamedfs.namenode.handler.count/namevalue10/value
/propertydfs.namenode.handler.count20×〖log〗_e^(Cluster Size)比如集群规模为8台时此参数设置为41。可通过简单的python代码计算该值代码如下。
[atguiguhadoop102 ~]$ 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 mathprint int(20*math.log(8))
41quit()②YARN参数调优yarn-site.xml
情景描述总共7台机器每天几亿条数据数据源-Flume-Kafka-HDFS-Hive
面临问题数据统计主要用HiveSQL没有数据倾斜小文件已经做了合并处理开启的JVM重用而且IO没有阻塞内存用了不到50%。但是还是跑的非常慢而且数据量洪峰过来时整个集群都会宕掉。基于这种情况有没有优化方案。
解决办法
内存利用率不够。这个一般是Yarn的2个配置造成的单个任务可以申请的最大内存大小和Hadoop单个节点可用内存大小。调节这两个参数能提高系统内存的利用率。
ayarn.nodemanager.resource.memory-mb
表示该节点上YARN可使用的物理内存总量默认是8192MB注意如果你的节点内存资源不够8GB则需要调减小这个值而YARN不会智能的探测节点的物理内存总量。
byarn.scheduler.maximum-allocation-mb
单个任务可申请的最多物理内存量默认是8192MB。
4.2.3 Zookeeper安装
1安装步骤
详见尚硅谷大数据技术之Zookeeper
4.2.4 Kafka安装
1安装步骤
详见尚硅谷大数据技术之Kafka
4.2.5 Flume安装
按照采集通道规划需在hadoop102hadoop103hadoop104三台节点分别部署一个Flume。可参照以下步骤先在hadoop102安装然后再进行分发。
1安装步骤
详见尚硅谷大数据技术之Flume
2分发Flume
[atguiguhadoop102 ~]$ xsync /opt/module/flume/3项目经验
1堆内存调整
Flume堆内存通常设置为4G或更高配置方式如下
修改/opt/module/flume/conf/flume-env.sh文件配置如下参数**虚拟机环境暂不配置**
export JAVA_OPTS-Xms4096m -Xmx4096m -Dcom.sun.management.jmxremote注
-Xms表示JVM Heap(堆内存)最小尺寸初始分配
-Xmx 表示JVM Heap(堆内存)最大允许的尺寸按需分配。
4.3 日志采集Flume
4.3.1 日志采集Flume配置概述
按照规划需要采集的用户行为日志文件分布在hadoop102hadoop103两台日志服务器故需要在hadoop102hadoop103两台节点配置日志采集Flume。日志采集Flume需要采集日志文件内容并对日志格式JSON进行校验然后将校验通过的日志发送到Kafka。
此处可选择TaildirSource和KafkaChannel并配置日志校验拦截器。
选择TailDirSource和KafkaChannel的原因如下
1TailDirSource
TailDirSource相比ExecSource、SpoolingDirectorySource的优势
TailDirSource断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置实现断点续传。
ExecSource可以实时搜集数据但是在Flume不运行或者Shell命令出错的情况下数据将会丢失。
SpoolingDirectorySource监控目录支持断点续传。
2KafkaChannel
采用Kafka Channel省去了Sink提高了效率。
日志采集Flume关键配置如下 4.3.2 日志采集Flume配置实操
1创建Flume配置文件
在hadoop102节点的Flume的job目录下创建file_to_kafka.conf
[atguiguhadoop102 flume]$ mkdir job
[atguiguhadoop102 flume]$ vim job/file_to_kafka.conf2配置文件内容如下
#定义组件
a1.sources r1
a1.channels c1#配置source
a1.sources.r1.type TAILDIR
a1.sources.r1.filegroups f1
a1.sources.r1.filegroups.f1 /opt/module/applog/log/app.*
a1.sources.r1.positionFile /opt/module/flume/taildir_position.json
a1.sources.r1.interceptors i1
a1.sources.r1.interceptors.i1.type com.atguigu.gmall.flume.interceptor.ETLInterceptor$Builder#配置channel
a1.channels.c1.type org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.channels.c1.kafka.topic topic_log
a1.channels.c1.parseAsFlumeEvent false#组装
a1.sources.r1.channels c13编写Flume拦截器
1创建Maven工程flume-interceptor
2创建包com.atguigu.gmall.flume.interceptor
3在pom.xml文件中添加如下配置
dependenciesdependencygroupIdorg.apache.flume/groupIdartifactIdflume-ng-core/artifactIdversion1.9.0/versionscopeprovided/scope/dependencydependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.62/version/dependency
/dependenciesbuildpluginspluginartifactIdmaven-compiler-plugin/artifactIdversion2.3.2/versionconfigurationsource1.8/sourcetarget1.8/target/configuration/pluginpluginartifactIdmaven-assembly-plugin/artifactIdconfigurationdescriptorRefsdescriptorRefjar-with-dependencies/descriptorRef/descriptorRefs/configurationexecutionsexecutionidmake-assembly/idphasepackage/phasegoalsgoalsingle/goal/goals/execution/executions/plugin/plugins
/build4在com.atguigu.gmall.flume.utils包下创建JSONUtil类
package com.atguigu.gmall.flume.utils;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;}}
}5在com.atguigu.gmall.flume.interceptor包下创建ETLInterceptor类
package com.atguigu.gmall.flume.interceptor;import com.atguigu.gmall.flume.utils.JSONUtil;
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 {Overridepublic void initialize() {}Overridepublic Event intercept(Event event) {//1、获取body当中的数据并转成字符串byte[] body event.getBody();String log new String(body, StandardCharsets.UTF_8);//2、判断字符串是否是一个合法的json是返回当前event不是返回nullif (JSONUtil.isJSONValidate(log)) {return event;} else {return null;}}Overridepublic ListEvent intercept(ListEvent list) {IteratorEvent 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{Overridepublic Interceptor build() {return new ETLInterceptor();}Overridepublic void configure(Context context) {}}Overridepublic void close() {}
}6打包 7需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面。
4.3.3 日志采集Flume测试
1启动Zookeeper、Kafka集群
2启动hadoop102的日志采集Flume
[atguiguhadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.loggerinfo,console3启动一个Kafka的Console-Consumer
[atguiguhadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_log4生成模拟数据
[atguiguhadoop102 ~]$ lg.sh5观察Kafka消费者是否能消费到数据
4.3.4 日志采集Flume启停脚本
1分发日志采集Flume配置文件和拦截器
若上述测试通过需将hadoop102节点的Flume的配置文件和拦截器jar包向另一台日志服务器发送一份。
[atguiguhadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/
[atguiguhadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/2方便起见此处编写一个日志采集Flume进程的启停脚本
在hadoop102节点的/home/atguigu/bin目录下创建脚本f1.sh
[atguiguhadoop102 bin]$ vim f1.sh在脚本中填写如下内容
#!/bin/bashcase $1 in
start){for i in hadoop102 hadoop103doecho --------启动 $i 采集flume-------ssh $i nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf /dev/null 21 done
};;
stop){for i in hadoop102 hadoop103doecho --------停止 $i 采集flume-------ssh $i ps -ef | grep file_to_kafka | grep -v grep |awk {print \$2} | xargs -n1 kill -9 done};;
esac3增加脚本执行权限
[atguiguhadoop102 bin]$ chmod 777 f1.sh4f1启动
[atguiguhadoop102 module]$ f1.sh start5f2停止
[atguiguhadoop102 module]$ f1.sh stop
