Hadoop如何安裝配置

這篇文章主要介紹“Hadoop如何安裝配置”，在日常操作中，相信很多人在Hadoop如何安裝配置問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”Hadoop如何安裝配置”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

4、數(shù)據(jù)采集模塊

4.1 集群所有進(jìn)程查看腳本

1）在/home/atguigu/bin目錄下創(chuàng)建腳本xcall.sh
        vim xcall.sh

2）在腳本中編寫(xiě)如下內(nèi)容

#! /bin/bash
#1、判斷是否輸入?yún)?shù)
if [ $# -lt 1 ]
then
	echo "必須輸入至少一個(gè)參數(shù)..."
	exit
fi
#2、執(zhí)行命令
for host in hadoop102 hadoop103 hadoop104
do
	echo "===================$host=================="
	ssh $host "$*"
done
#xcall.sh mkdir -p /xx/xx

3）修改腳本執(zhí)行權(quán)限
        chmod +x xcall.sh

4）啟動(dòng)腳本
        xcall.sh jps

4.2 Hadoop安裝

安裝Hadoop

4.2.1 項(xiàng)目經(jīng)驗(yàn)之HDFS存儲(chǔ)多目錄

1）生產(chǎn)環(huán)境服務(wù)器磁盤(pán)情況

2）在hdfs-site.xml文件中配置多目錄，注意新掛載磁盤(pán)的訪問(wèn)權(quán)限問(wèn)題。
        HDFS的DataNode節(jié)點(diǎn)保存數(shù)據(jù)的路徑由dfs.datanode.data.dir參數(shù)決定，其默認(rèn)值為file://${hadoop.tmp.dir}/dfs/data，若服務(wù)器有多個(gè)磁盤(pán)，必須對(duì)該參數(shù)進(jìn)行修改。如服務(wù)器磁盤(pán)如上圖所示，則該參數(shù)應(yīng)修改為如下的值。

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

注意：每臺(tái)服務(wù)器掛載的磁盤(pán)不一樣，所以每個(gè)節(jié)點(diǎn)的多目錄配置可以不一致。單獨(dú)配置即可。

總結(jié)：
1、HDFS存儲(chǔ)多目錄
	1、好處:
		1、增大存儲(chǔ)容量
		2、提高IO、并發(fā)
	2、如何實(shí)現(xiàn)存儲(chǔ)多目錄:
		需要在hdfs-site.xml中配置
		<property>
			<name>dfs.datanode.data.dir</name>
			<value>file://磁盤(pán)掛載點(diǎn)1/存儲(chǔ)路徑,file://磁盤(pán)掛載點(diǎn)2/存儲(chǔ)路徑</value>
		</property>
在真實(shí)的生產(chǎn)環(huán)境中,每天服務(wù)器磁盤(pán)數(shù)以及掛載點(diǎn)可能都不太一樣,所以該配置需要在每臺(tái)服務(wù)器上單獨(dú)配置

4.2.2 集群數(shù)據(jù)均衡

   數(shù)據(jù)可能累積在某個(gè)或者某個(gè)datanode節(jié)點(diǎn)上,或者服務(wù)器有多塊磁盤(pán)的時(shí)候,數(shù)據(jù)累積在某一塊磁盤(pán)上,所以此時(shí)可能會(huì)造成某些
節(jié)點(diǎn)或者磁盤(pán)負(fù)載壓力會(huì)比較大,所以需要在節(jié)點(diǎn)以及磁盤(pán)上進(jìn)行數(shù)據(jù)均衡
	1、節(jié)點(diǎn)數(shù)據(jù)均衡
		1、開(kāi)啟節(jié)點(diǎn)數(shù)據(jù)均衡: start-balancer --threshould N
			N代表節(jié)點(diǎn)之間磁盤(pán)的利用率不能超過(guò)N%
		2、停止節(jié)點(diǎn)數(shù)據(jù)均衡: stop-balancer
	2、磁盤(pán)的數(shù)據(jù)均衡
		1、生成執(zhí)行計(jì)劃: hdfs diskbalancer -plan 主機(jī)名 <代表指定主機(jī)需要執(zhí)行磁盤(pán)的數(shù)據(jù)均衡>
		2、開(kāi)啟均衡: hdfs diskbalancer -execute 主機(jī).plan.json
		3、查看均衡進(jìn)度: hdfs diskbalancer -query 主機(jī)名
		4、停止均衡: hdfs diskbalancer -cancel 主機(jī).plan.json
不管是節(jié)點(diǎn)數(shù)據(jù)均衡還是磁盤(pán)數(shù)據(jù)均衡都需要選擇集群空閑的時(shí)候進(jìn)行,因?yàn)榫?nbsp;的時(shí)候需要消耗大量的磁盤(pán)IO以及網(wǎng)絡(luò)IO。

4.2.3 項(xiàng)目經(jīng)驗(yàn)之支持LZO壓縮配置

hadoop本身并不支持lzo壓縮，如果想要支持lzo壓縮，需要額外操作配置。
1）將編譯好后的hadoop-lzo-0.4.20.jar 放入hadoop-3.1.3/share/hadoop/common/

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

2）同步hadoop-lzo-0.4.20.jar到hadoop103、hadoop104

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

3）core-site.xml增加配置支持LZO壓縮

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

com.hadoop.compression.lzo.LzoCodec: 后續(xù)MR讀取lzo壓縮文件的時(shí)候不會(huì)切片 com.hadoop.compression.lzo.LzopCodec: 后續(xù)MR讀取lzo壓縮文件的時(shí)候會(huì)切片<前提是生成索引文件>

4）同步core-site.xml到hadoop103、hadoop104

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

5）啟動(dòng)及查看集群

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

4.2.4 項(xiàng)目經(jīng)驗(yàn)之LZO創(chuàng)建索引

1）創(chuàng)建LZO文件的索引，LZO壓縮文件的可切片特性依賴于其索引，故我們需要手動(dòng)為L(zhǎng)ZO壓縮文件創(chuàng)建索引。若無(wú)索引，則LZO文件的切片只有一個(gè)。

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

4.2.5 項(xiàng)目經(jīng)驗(yàn)之基準(zhǔn)測(cè)試

1、HDFS吞吐量測(cè)試
①、寫(xiě)吞吐量: 
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個(gè)數(shù)應(yīng)該設(shè)置為集群所有CPU個(gè)數(shù)-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
	測(cè)試結(jié)果:
	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 <單個(gè)maptask寫(xiě)的速率>
	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讀/寫(xiě)的吞吐量 = 單個(gè)maptask寫(xiě)的速率 * nrFiles
③、刪除測(cè)試生成數(shù)據(jù)
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.1.3-tests.jar TestDFSIO -clean

2、MR計(jì)算性能測(cè)試:<使用Sort程序評(píng)測(cè)MapReduce>

（1）使用RandomWriter來(lái)產(chǎn)生隨機(jī)數(shù)，每個(gè)節(jié)點(diǎn)運(yùn)行10個(gè)Map任務(wù)，每個(gè)Map產(chǎn)生大約1G大小的二進(jìn)制隨機(jī)數(shù)
hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.1.3.jar randomwriter random-data

（2）執(zhí)行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）驗(yàn)證數(shù)據(jù)是否真正排好序了
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

4.2.6 項(xiàng)目經(jīng)驗(yàn)之Hadoop參數(shù)調(diào)優(yōu)

1、namenode線程池的線程數(shù)調(diào)整:
    原因:namenode中有一個(gè)線程池,線程池中的線程主要用于datanode與namenode心跳以及客戶端獲取/寫(xiě)入元數(shù)據(jù)的請(qǐng)求，所以如
果并發(fā)比較大或者datanode比較多,就可能導(dǎo)致線程池中線程不夠用。通常需要增大參數(shù)dfs.namenode.handler.count的默認(rèn)值10
如何調(diào)整：
	在hdfs-site.xml中配置:
	<property>
		<name>dfs.namenode.handler.count</name>
		<value>10</value>
	</property>
調(diào)整為多少: 20 * log(集群機(jī)器臺(tái)數(shù)) <以e為底>

2、調(diào)整nodemanager能夠使用的資源數(shù)
    原因: 服務(wù)器的資源不會(huì)全部給nodemanager使用,默認(rèn)提供給nodemanager使用的內(nèi)存8G,在真實(shí)生產(chǎn)中該默認(rèn)值不夠用需要進(jìn)
行調(diào)整
如何調(diào)整:
	在yarn-site.xml中配置:
	<property>
		<name>yarn.nodemanager.resource.memory-mb</name>
		<value>1024</value>
	</property>
一般設(shè)置為服務(wù)器總內(nèi)存的70%-80%

4.3 Zookeeper安裝

4.3.1 安裝ZK

安裝Zookeeper

4.3.2 ZK集群?jiǎn)?dòng)停止腳本

#! /bin/bash
#1、判斷參數(shù)是否傳入
if [ $# -lt 1 ]
then
	echo "必須傳入?yún)?shù)..."
	exit
fi
#2、根據(jù)參數(shù)執(zhí)行對(duì)應(yīng)的邏輯
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 "參數(shù)輸入錯(cuò)誤..."
;;
esac

zookeeper常用指令
	1、創(chuàng)建節(jié)點(diǎn): create 節(jié)點(diǎn)路徑
	2、保存數(shù)據(jù): set 路徑 數(shù)據(jù)
	3、獲取數(shù)據(jù): get 路徑
	4、查看節(jié)點(diǎn): ls 路徑
	5、刪除節(jié)點(diǎn):
		1、刪除非空節(jié)點(diǎn): deleteall 路徑
		2、刪除空節(jié)點(diǎn): delete 路徑

4.4 Kafka安裝

4.4.1 Kafka集群安裝

安裝Kafka

4.4.2 Kafka集群?jiǎn)?dòng)停止腳本

#! /bin/bash
#1、判斷參數(shù)是否傳入
if [ $# -lt 1 ]
then
	echo "必須傳入?yún)?shù)..."
	exit
fi
#2、根據(jù)參數(shù)匹配邏輯
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進(jìn)程正常" || echo "kafka進(jìn)程不存在"
	done
;;
*)
	echo "參數(shù)輸入錯(cuò)誤...."
;;
esac

4.4.3 Kafka常用命令

1、shell常用命令
   1、topic相關(guān)
      1、創(chuàng)建topic: bin/kafka-topics.sh --create --topic topic名稱 --bootstrap-server hadoop102:9092,hadoop103:9092 --partitions 分區(qū)數(shù) --replication-factor 副本數(shù)
      2、查看集群所有topic: bin/kafka-topics.sh --list --bootstrap-server hadoop102:9092
      3、查看topic詳細(xì)信息: bin/kafka-topics.sh --describe --topic topic名稱 --bootstrap-server hadoop102:9092
      4、修改topic<只能修改分區(qū),而且只能增加分區(qū)>: bin/kafka-topics.sh --alter --topic topic名稱 --bootstrap-server hadoop102:9092 --partitions 分區(qū)數(shù)
      5、刪除topic: bin/kafka-topics.sh --delete --topic topic名稱 --bootstrap-server hadoop102:9092

   2、生產(chǎn)者相關(guān): bin/kafka-console-producer.sh --topic topic名稱 --broker-list hadoop102:9092
   3、消費(fèi)者相關(guān): 
      1、消費(fèi)數(shù)據(jù):bin/kafka-console-consumer.sh --topic topic名稱 --bootstrap-server hadoop102:9092 [--group 消費(fèi)者id] [--from-beginning]
      2、查看消費(fèi)者組消費(fèi)topic的進(jìn)度: bin/kafka-consumer-group.sh --all-groups --all-topics --describe --bootstrap-server hadoop102:9092

   4、數(shù)據(jù)相關(guān): bin/kafka-dump-log.sh --files 待查看文件路徑 --print-data-log

4.4.4 項(xiàng)目經(jīng)驗(yàn)之Kafka壓力測(cè)試

1、kafka吞吐量測(cè)試:
   1、讀吞吐量: 
   bin/kafka-consumer-perf-test.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic topic名稱 --messages 總共拉取多少條數(shù)據(jù) 
      測(cè)試結(jié)果:
      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
      開(kāi)始測(cè)試時(shí)間，測(cè)試結(jié)束數(shù)據(jù)，共消費(fèi)數(shù)據(jù)9.5368MB，吞吐量2.0714MB/s，共消費(fèi)100010條，平均每秒消費(fèi)21722.4153條。
      MB.sec就是吞吐量測(cè)試結(jié)果

   2、寫(xiě)吞吐量：
   bin/kafka-producer-perf-test.sh --record-size 指定每條數(shù)據(jù)大小 --topic topic名稱 --num-records 寫(xiě)入topic數(shù)據(jù)條數(shù) --producer-props bootstrap.servers=hadoop102:9092 --throughput 寫(xiě)入的速率[-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就是吞吐量測(cè)試結(jié)果

4.4.5 項(xiàng)目經(jīng)驗(yàn)之Kafka機(jī)器數(shù)量計(jì)算

Kafka機(jī)器數(shù)量（經(jīng)驗(yàn)公式）=2*（峰值生產(chǎn)速度*副本數(shù)/100）+1
先拿到峰值生產(chǎn)速度，再根據(jù)設(shè)定的副本數(shù)，就能預(yù)估出需要部署Kafka的數(shù)量。
比如我們的峰值生產(chǎn)速度是50M/s。副本數(shù)為2。
Kafka機(jī)器數(shù)量=2*（50*2/100）+ 1=3臺(tái)

4.4.6 項(xiàng)目經(jīng)驗(yàn)之Kafka分區(qū)數(shù)計(jì)算

1）創(chuàng)建一個(gè)只有1個(gè)分區(qū)的topic
2）測(cè)試這個(gè)topic的producer吞吐量和consumer吞吐量。
3）假設(shè)他們的值分別是Tp和Tc，單位可以是MB/s。
4）然后假設(shè)總的目標(biāo)吞吐量是Tt，那么分區(qū)數(shù)=Tt / min（Tp，Tc）
例如：producer吞吐量=20m/s；consumer吞吐量=50m/s，期望吞吐量100m/s；
分區(qū)數(shù)=100 / 20 =5分區(qū)
https://blog.csdn.net/weixin_42641909/article/details/89294698
分區(qū)數(shù)一般設(shè)置為：3-10個(gè)

4.5 采集日志Flume

4.5.1 日志采集Flume安裝

安裝三臺(tái)Flume（日志采集：102、103，消費(fèi)kafka：104）

4.5.2 項(xiàng)目經(jīng)驗(yàn)之Flume組件選型

tailDirSource，kafkaChannel

4.5.3 日志采集Flume配置

（1）在/opt/module/flume/conf目錄下創(chuàng)建file-flume-kafka.conf文件
[atguigu@hadoop102 conf]$ vim file-flume-kafka.conf
在文件配置如下內(nèi)容

#1、定義agent、source、channel名稱
a1.sources = r1
a1.channels = c1

#2、描述source<taildir>
a1.sources.r1.type = TAILDIR
#指定文件組的名稱
a1.sources.r1.filegroups = f1
#指定組監(jiān)控的目錄
a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
#指定斷點(diǎn)續(xù)傳文件
a1.sources.r1.positionFile = /opt/module/flume/position.json
#指定一個(gè)批次采集多少數(shù)據(jù)
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
#指定數(shù)據(jù)寫(xiě)入的topic名稱
a1.channels.c1.kafka.topic = applog
#數(shù)據(jù)寫(xiě)入kafka的時(shí)候是否以event格式寫(xiě)入: true=是 false: 不是,只寫(xiě)body數(shù)據(jù)
a1.channels.c1.parseAsFlumeEvent = false

#5、關(guān)聯(lián)source->channel
a1.sources.r1.channels = c1

4.5.4 Flume欄截器

 @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;
    }

4.5.5 日志采集Flume啟動(dòng)停止腳本

#! /bin/bash
#1、判斷參數(shù)是否傳入
if [ $# -lt 1 ]
then
	echo "必須傳入?yún)?shù)...."
fi
#2、根據(jù)參數(shù)匹配邏輯
case $1 in
"start")
	for host in hadoop102 hadoop103
	do
		echo "===============開(kāi)啟$host服務(wù)器采集==============="
		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服務(wù)器采集==============="
		ssh $host "ps -ef |grep file_to_kafka.conf | grep -v grep | awk '{print \$2}'| xargs kill -9"
	done
;;
*)
	echo "參數(shù)輸入錯(cuò)誤..."
;;
esac

4.6 消費(fèi)Kafka數(shù)據(jù)Flume

4.6.1項(xiàng)目經(jīng)驗(yàn)之Flume逐漸選型

kafkaSource、fileChannel/memoryChannel、hdfsSink

4.6.2 Flume蘭截器

    @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;
    }

4.6.3 日志消峰Flume配置

#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
#指定從哪個(gè)topic讀取數(shù)據(jù)
a1.sources.r1.kafka.topics = applog
#指定消費(fèi)者組的id
a1.sources.r1.kafka.consumer.group.id = g1
#指定source從kafka一個(gè)批次拉取多少條消息: batchSize<=事務(wù)容量<=channel容量
a1.sources.r1.batchSize = 100
#指定消費(fèi)者組第一個(gè)消費(fèi)topic的數(shù)據(jù)的時(shí)候從哪里開(kāi)始消費(fèi)
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
#指定數(shù)據(jù)保存在本地磁盤(pán)哪個(gè)目錄
a1.channels.c1.dataDirs = /opt/module/flume/datas
#指定channel內(nèi)存中event的指針數(shù)據(jù)
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint
#指定checkpoint的持久化的間隔時(shí)間
a1.channels.c1.checkpointInterval = 5000
#指定channel容量
a1.channels.c1.capacity = 1000000

#5、描述sink
a1.sinks.k1.type = hdfs
#指定數(shù)據(jù)存儲(chǔ)目錄
a1.sinks.k1.hdfs.path = hdfs://hadoop102:8020/applog/%Y%m%d
#指定文件的前綴
a1.sinks.k1.hdfs.filePrefix = log-
#指定滾動(dòng)生成文件的時(shí)間間隔
a1.sinks.k1.hdfs.rollInterval = 30
#指定滾動(dòng)生成文件的大小
a1.sinks.k1.hdfs.rollSize = 132120576
#指定寫(xiě)入多少個(gè)event之后滾動(dòng)生成新文件<一般禁用>
a1.sinks.k1.hdfs.rollCount = 0
#指定sink每個(gè)批次從channel拉取多少數(shù)據(jù)
a1.sinks.k1.hdfs.batchSize = 100
#指定寫(xiě)入hdfs的時(shí)候壓縮格式
#a1.sinks.k1.hdfs.codeC = lzop
#指定文件寫(xiě)入的是格式 SequenceFile-序列化文件, DataStream-文本文件, CompressedStream-壓縮文件
#a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.fileType = DataStream

#6、關(guān)聯(lián)source->channel->sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

4.6.4 日志消費(fèi)Flume啟動(dòng)停止腳本

1）在/home/atguigu/bin目錄下創(chuàng)建腳本f2.sh
[atguigu@hadoop102 bin]$ vim f2.sh
在腳本中填寫(xiě)如下內(nèi)容

#! /bin/bash

case $1 in
"start"){
        for i in hadoop104
        do
                echo " --------啟動(dòng) $i 消費(fè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 消費(fèi)flume-------"
                ssh $i "ps -ef | grep kafka-flume-hdfs | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
        done

};;
esac

2）增加腳本執(zhí)行權(quán)限
[atguigu@hadoop102 bin]$ chmod u+x f2.sh

3）f2集群?jiǎn)?dòng)腳本
[atguigu@hadoop102 module]$ f2.sh start

4）f2集群停止腳本
[atguigu@hadoop102 module]$ f2.sh stop

4.6.5 項(xiàng)目經(jīng)驗(yàn)之Flume內(nèi)存優(yōu)化

flume默認(rèn)的內(nèi)存最大是2000M,一般在項(xiàng)目中需要設(shè)置為4G左右
	Xms: 啟動(dòng)內(nèi)存大小
	Xmx: 運(yùn)行時(shí)最大內(nèi)存大小
Xms與Xmx在設(shè)置的時(shí)候最好設(shè)置的一樣。因?yàn)樵O(shè)置的不一樣,啟動(dòng)時(shí)Xms內(nèi)存比較小,導(dǎo)致很快內(nèi)存不夠用,需要進(jìn)行內(nèi)存的擴(kuò)容<在擴(kuò)容
之前會(huì)進(jìn)行GC>,GC會(huì)影響性能。兩者設(shè)置為一樣就不會(huì)出現(xiàn)內(nèi)存擴(kuò)容,減少GC次數(shù)

到此，關(guān)于“Hadoop如何安裝配置”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！