Hadoop如何安装配置

#! /bin/bash
#1、判断是否输入参数
if [ $# -lt 1 ]
then
	echo "必须输入至少一个参数..."
	exit
fi
#2、执行命令
for host in hadoop102 hadoop103 hadoop104
do
	echo "===================$host=================="
	ssh $host "$*"
done
#xcall.sh mkdir -p /xx/xx

<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>

总结：
1、HDFS存储多目录
	1、好处:
		1、增大存储容量
		2、提高IO、并发
	2、如何实现存储多目录:
		需要在hdfs-site.xml中配置
		<property>
			<name>dfs.datanode.data.dir</name>
			<value>file://磁盘挂载点1/存储路径,file://磁盘挂载点2/存储路径</value>
		</property>
在真实的生产环境中,每天服务器磁盘数以及挂载点可能都不太一样,所以该配置需要在每台服务器上单独配置

   数据可能累积在某个或者某个datanode节点上,或者服务器有多块磁盘的时候,数据累积在某一块磁盘上,所以此时可能会造成某些
节点或者磁盘负载压力会比较大,所以需要在节点以及磁盘上进行数据均衡
	1、节点数据均衡
		1、开启节点数据均衡: start-balancer --threshould N
			N代表节点之间磁盘的利用率不能超过N%
		2、停止节点数据均衡: stop-balancer
	2、磁盘的数据均衡
		1、生成执行计划: hdfs diskbalancer -plan 主机名 <代表指定主机需要执行磁盘的数据均衡>
		2、开启均衡: hdfs diskbalancer -execute 主机.plan.json
		3、查看均衡进度: hdfs diskbalancer -query 主机名
		4、停止均衡: hdfs diskbalancer -cancel 主机.plan.json
不管是节点数据均衡还是磁盘数据均衡都需要选择集群空闲的时候进行,因为均衡 的时候需要消耗大量的磁盘IO以及网络IO。

[atguigu@hadoop102 common]$ pwd
/opt/module/hadoop-3.1.3/share/hadoop/common
[atguigu@hadoop102 common]$ ls
hadoop-lzo-0.4.20.jar

[atguigu@hadoop102 common]$ xsync hadoop-lzo-0.4.20.jar

    <property>
        <name>io.compression.codecs</name>
        <value>
            org.apache.hadoop.io.compress.GzipCodec,
            org.apache.hadoop.io.compress.DefaultCodec,
            org.apache.hadoop.io.compress.BZip2Codec,
            org.apache.hadoop.io.compress.SnappyCodec,
            com.hadoop.compression.lzo.LzoCodec,
            com.hadoop.compression.lzo.LzopCodec
        </value>
    </property>

    <property>
        <name>io.compression.codec.lzo.class</name>
        <value>com.hadoop.compression.lzo.LzoCodec</value>
    </property>

[atguigu@hadoop102 hadoop]$ xsync core-site.xml

[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh

hadoop jar /opt/module/hadoop/share/hadoop/common/hadoop-lzo-4.1.x.jar com.hadoop.compression.lzo.DistributedLzoIndexer 需要建索引的lzo文件名

1、HDFS吞吐量测试
①、写吞吐量: 
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.1.3-tests.jar TestDFSIO -write -nrFiles N -size 128MB
//-nrFiles个数应该设置为集群所有CPU个数-1

②、读吞吐量:
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.1.3-tests.jar TestDFSIO -read -nrFiles N -size 128MB
	测试结果:
	2020-04-16 13:41:24,724 INFO fs.TestDFSIO: ----- TestDFSIO ----- : write
	2020-04-16 13:41:24,724 INFO fs.TestDFSIO:             Date & time: Thu Apr 16 13:41:24 CST 2020
	2020-04-16 13:41:24,724 INFO fs.TestDFSIO:         Number of files: 10
	2020-04-16 13:41:24,725 INFO fs.TestDFSIO:  Total MBytes processed: 1280
	2020-04-16 13:41:24,725 INFO fs.TestDFSIO:       Throughput mb/sec: 8.88 <单个maptask写的速率>
	2020-04-16 13:41:24,725 INFO fs.TestDFSIO:  Average IO rate mb/sec: 8.96
	2020-04-16 13:41:24,725 INFO fs.TestDFSIO:   IO rate std deviation: 0.87
	2020-04-16 13:41:24,725 INFO fs.TestDFSIO:      Test exec time sec: 67.61

HDFS读/写的吞吐量 = 单个maptask写的速率 * nrFiles
③、删除测试生成数据
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.1.3-tests.jar TestDFSIO -clean

2、MR计算性能测试:<使用Sort程序评测MapReduce>

（1）使用RandomWriter来产生随机数，每个节点运行10个Map任务，每个Map产生大约1G大小的二进制随机数
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar randomwriter random-data

（2）执行Sort程序
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar sort random-data sorted-data

（3）验证数据是否真正排好序了
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.1.3-tests.jar testmapredsort -sortInput random-data -sortOutput sorted-data

1、namenode线程池的线程数调整:
    原因:namenode中有一个线程池,线程池中的线程主要用于datanode与namenode心跳以及客户端获取/写入元数据的请求，所以如
果并发比较大或者datanode比较多,就可能导致线程池中线程不够用。通常需要增大参数dfs.namenode.handler.count的默认值10
如何调整：
	在hdfs-site.xml中配置:
	<property>
		<name>dfs.namenode.handler.count</name>
		<value>10</value>
	</property>
调整为多少: 20 * log(集群机器台数) <以e为底>

2、调整nodemanager能够使用的资源数
    原因: 服务器的资源不会全部给nodemanager使用,默认提供给nodemanager使用的内存8G,在真实生产中该默认值不够用需要进
行调整
如何调整:
	在yarn-site.xml中配置:
	<property>
		<name>yarn.nodemanager.resource.memory-mb</name>
		<value>1024</value>
	</property>
一般设置为服务器总内存的70%-80%

#! /bin/bash
#1、判断参数是否传入
if [ $# -lt 1 ]
then
	echo "必须传入参数..."
	exit
fi
#2、根据参数执行对应的逻辑
case $1 in
"start")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=======================start $host zookeeper==============="
		ssh $host "/opt/module/zookeeper/bin/zkServer.sh start"
	done
;;
"stop")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=======================stop $host zookeeper==============="
		ssh $host "/opt/module/zookeeper/bin/zkServer.sh stop"
	done
;;
"status")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=======================status $host zookeeper==============="
		ssh $host "/opt/module/zookeeper/bin/zkServer.sh status"
	done
;;
*)
	echo "参数输入错误..."
;;
esac

zookeeper常用指令
	1、创建节点: create 节点路径
	2、保存数据: set 路径 数据
	3、获取数据: get 路径
	4、查看节点: ls 路径
	5、删除节点:
		1、删除非空节点: deleteall 路径
		2、删除空节点: delete 路径

#! /bin/bash
#1、判断参数是否传入
if [ $# -lt 1 ]
then
	echo "必须传入参数..."
	exit
fi
#2、根据参数匹配逻辑
case $1 in
"start")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=================start $host kafka=============="
		ssh $host "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
	done
;;
"stop")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=================stop $host kafka=============="
		ssh $host "/opt/module/kafka/bin/kafka-server-stop.sh"
	done
;;
"status")
	for host in hadoop102 hadoop103 hadoop104
	do
		echo "=================status $host kafka=============="
		pid=$(ssh $host "ps -ef | grep server.properties | grep -v grep")
		[ "$pid" ] && echo "kafka进程正常" || echo "kafka进程不存在"
	done
;;
*)
	echo "参数输入错误...."
;;
esac

1、shell常用命令
   1、topic相关
      1、创建topic: bin/kafka-topics.sh --create --topic topic名称 --bootstrap-server hadoop102:9092,hadoop103:9092 --partitions 分区数 --replication-factor 副本数
      2、查看集群所有topic: bin/kafka-topics.sh --list --bootstrap-server hadoop102:9092
      3、查看topic详细信息: bin/kafka-topics.sh --describe --topic topic名称 --bootstrap-server hadoop102:9092
      4、修改topic<只能修改分区,而且只能增加分区>: bin/kafka-topics.sh --alter --topic topic名称 --bootstrap-server hadoop102:9092 --partitions 分区数
      5、删除topic: bin/kafka-topics.sh --delete --topic topic名称 --bootstrap-server hadoop102:9092

   2、生产者相关: bin/kafka-console-producer.sh --topic topic名称 --broker-list hadoop102:9092
   3、消费者相关: 
      1、消费数据:bin/kafka-console-consumer.sh --topic topic名称 --bootstrap-server hadoop102:9092 [--group 消费者id] [--from-beginning]
      2、查看消费者组消费topic的进度: bin/kafka-consumer-group.sh --all-groups --all-topics --describe --bootstrap-server hadoop102:9092

   4、数据相关: bin/kafka-dump-log.sh --files 待查看文件路径 --print-data-log

1、kafka吞吐量测试:
   1、读吞吐量: 
   bin/kafka-consumer-perf-test.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic topic名称 --messages 总共拉取多少条数据 
      测试结果:
      start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec
      2019-02-19 20:29:07:566, 2019-02-19 20:29:12:170, 9.5368, 2.0714, 100010, 21722.4153
      开始测试时间，测试结束数据，共消费数据9.5368MB，吞吐量2.0714MB/s，共消费100010条，平均每秒消费21722.4153条。
      MB.sec就是吞吐量测试结果

   2、写吞吐量：
   bin/kafka-producer-perf-test.sh --record-size 指定每条数据大小 --topic topic名称 --num-records 写入topic数据条数 --producer-props bootstrap.servers=hadoop102:9092 --throughput 写入的速率[-1代表不限制]
      100000 records sent, 95877.277085 records/sec (9.14 MB/sec), 187.68 ms avg latency, 424.00 ms max latency, 155 ms 50th, 411 ms 95th, 423 ms 99th, 424 ms 99.9th.
      9.14 MB/sec就是吞吐量测试结果

Kafka机器数量（经验公式）=2*（峰值生产速度*副本数/100）+1
先拿到峰值生产速度，再根据设定的副本数，就能预估出需要部署Kafka的数量。
比如我们的峰值生产速度是50M/s。副本数为2。
Kafka机器数量=2*（50*2/100）+ 1=3台

1）创建一个只有1个分区的topic
2）测试这个topic的producer吞吐量和consumer吞吐量。
3）假设他们的值分别是Tp和Tc，单位可以是MB/s。
4）然后假设总的目标吞吐量是Tt，那么分区数=Tt / min（Tp，Tc）
例如：producer吞吐量=20m/s；consumer吞吐量=50m/s，期望吞吐量100m/s；
分区数=100 / 20 =5分区
https://blog.csdn.net/weixin_42641909/article/details/89294698
分区数一般设置为：3-10个

#1、定义agent、source、channel名称
a1.sources = r1
a1.channels = c1

#2、描述source<taildir>
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/position.json
#指定一个批次采集多少数据
a1.sources.r1.batchSize = 100

#3、描述lan截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.interceptor.ETLInterceptor$Builder

#4、描述channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
#指定kafka集群地址
a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
#指定数据写入的topic名称
a1.channels.c1.kafka.topic = applog
#数据写入kafka的时候是否以event格式写入: true=是 false: 不是,只写body数据
a1.channels.c1.parseAsFlumeEvent = false

#5、关联source->channel
a1.sources.r1.channels = c1

 @Override
    public Event intercept(Event event) {

        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);

        if (JSONUtils.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;
    }

#! /bin/bash
#1、判断参数是否传入
if [ $# -lt 1 ]
then
	echo "必须传入参数...."
fi
#2、根据参数匹配逻辑
case $1 in
"start")
	for host in hadoop102 hadoop103
	do
		echo "===============开启$host服务器采集==============="
		ssh $host "nohup /opt/module/flume-1.9.0/bin/flume-ng agent -n a1 -c /opt/module/flume-1.9.0/conf/ -f /opt/module/flume-1.9.0/jobs/file_to_kafka.conf -Dflume.root.logger=INFO,LOGFILE >/opt/module/flume-1.9.0/logs 2>&1 &"
	done
;;
"stop")
	for host in hadoop102 hadoop103
	do
		echo "===============停止$host服务器采集==============="
		ssh $host "ps -ef |grep file_to_kafka.conf | grep -v grep | awk '{print \$2}'| xargs kill -9"
	done
;;
*)
	echo "参数输入错误..."
;;
esac

    @Override
    public Event intercept(Event event) {

        Map<String, String> headers = event.getHeaders();
        String log = new String(event.getBody(), StandardCharsets.UTF_8);

        JSONObject jsonObject = JSONObject.parseObject(log);

        String ts = jsonObject.getString("ts");
        headers.put("timestamp", ts);

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        events.clear();
        for (Event event : list) {
            events.add(intercept(event));
        }

        return events;
    }

#1、定义agent、channel、source、sink名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1

#2、描述source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
#指定kafka集群地址
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
#指定从哪个topic读取数据
a1.sources.r1.kafka.topics = applog
#指定消费者组的id
a1.sources.r1.kafka.consumer.group.id = g1
#指定source从kafka一个批次拉取多少条消息: batchSize<=事务容量<=channel容量
a1.sources.r1.batchSize = 100
#指定消费者组第一个消费topic的数据的时候从哪里开始消费
a1.sources.r1.kafka.consumer.auto.offset.reset = earliest

#3、描述lan截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.interceptor.MyTimestampInterceptor$Builder

#4、描述channel
a1.channels.c1.type = file
#指定数据保存在本地磁盘哪个目录
a1.channels.c1.dataDirs = /opt/module/flume/datas
#指定channel内存中event的指针数据
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint
#指定checkpoint的持久化的间隔时间
a1.channels.c1.checkpointInterval = 5000
#指定channel容量
a1.channels.c1.capacity = 1000000

#5、描述sink
a1.sinks.k1.type = hdfs
#指定数据存储目录
a1.sinks.k1.hdfs.path = hdfs://hadoop102:8020/applog/%Y%m%d
#指定文件的前缀
a1.sinks.k1.hdfs.filePrefix = log-
#指定滚动生成文件的时间间隔
a1.sinks.k1.hdfs.rollInterval = 30
#指定滚动生成文件的大小
a1.sinks.k1.hdfs.rollSize = 132120576
#指定写入多少个event之后滚动生成新文件<一般禁用>
a1.sinks.k1.hdfs.rollCount = 0
#指定sink每个批次从channel拉取多少数据
a1.sinks.k1.hdfs.batchSize = 100
#指定写入hdfs的时候压缩格式
#a1.sinks.k1.hdfs.codeC = lzop
#指定文件写入的是格式 SequenceFile-序列化文件, DataStream-文本文件, CompressedStream-压缩文件
#a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.fileType = DataStream

#6、关联source->channel->sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

#! /bin/bash

case $1 in
"start"){
        for i in hadoop104
        do
                echo " --------启动 $i 消费flume-------"
                ssh $i "nohup /opt/module/flume/bin/flume-ng agent --conf-file /opt/module/flume/conf/kafka-flume-hdfs.conf --name a1 -Dflume.root.logger=INFO,LOGFILE >/opt/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

flume默认的内存最大是2000M,一般在项目中需要设置为4G左右
	Xms: 启动内存大小
	Xmx: 运行时最大内存大小
Xms与Xmx在设置的时候最好设置的一样。因为设置的不一样,启动时Xms内存比较小,导致很快内存不够用,需要进行内存的扩容<在扩容
之前会进行GC>,GC会影响性能。两者设置为一样就不会出现内存扩容,减少GC次数