相關(guān)文章

• 【數(shù)倉(cāng)】基本概念、知識(shí)普及、核心技術(shù)
• 【數(shù)倉(cāng)】數(shù)據(jù)分層概念以及相關(guān)邏輯
• 【數(shù)倉(cāng)】Hadoop軟件安裝及使用（集群配置）
• 【數(shù)倉(cāng)】Hadoop集群配置常用參數(shù)說(shuō)明
• 【數(shù)倉(cāng)】zookeeper軟件安裝及集群配置
• 【數(shù)倉(cāng)】kafka軟件安裝及集群配置
• 【數(shù)倉(cāng)】flume軟件安裝及配置
• 【數(shù)倉(cāng)】flume常見(jiàn)配置總結(jié)，以及示例

一、flume有什么作用

Apache Flume是一個(gè)分布式、可靠且可用的大數(shù)據(jù)日志采集、聚合和傳輸系統(tǒng)。它主要用于將大量的日志數(shù)據(jù)從不同的數(shù)據(jù)源收集起來(lái)，然后通過(guò)通道（Channel）進(jìn)行傳輸，最終將數(shù)據(jù)傳輸?shù)街付ǖ哪康牡?，如HDFS、HBase等。Flume具有高度可擴(kuò)展性、容錯(cuò)性和靈活性，可以適應(yīng)各種復(fù)雜的數(shù)據(jù)采集場(chǎng)景。

Flume的核心組件包括Source、Channel和Sink。Source負(fù)責(zé)從數(shù)據(jù)源中讀取數(shù)據(jù)，可以是文件、網(wǎng)絡(luò)套接字、消息隊(duì)列等；Channel是數(shù)據(jù)的緩沖區(qū)，用于在Source和Sink之間傳輸數(shù)據(jù)；Sink負(fù)責(zé)將數(shù)據(jù)寫(xiě)入目標(biāo)存儲(chǔ)系統(tǒng)，如HDFS、HBase、Kafka等。此外，F(xiàn)lume還支持多種類型的Source、Channel和Sink，用戶可以根據(jù)實(shí)際需求進(jìn)行選擇和配置。

Flume的主要作用是實(shí)現(xiàn)大規(guī)模數(shù)據(jù)采集和傳輸，實(shí)現(xiàn)數(shù)據(jù)的實(shí)時(shí)處理和分析，從而為企業(yè)提供更好的業(yè)務(wù)決策支持。在實(shí)際應(yīng)用中，F(xiàn)lume可以用于日志收集、事件跟蹤、數(shù)據(jù)流處理等場(chǎng)景。通過(guò)將數(shù)據(jù)從不同的數(shù)據(jù)源采集并傳輸?shù)街付ǖ哪康牡?，F(xiàn)lume可以幫助企業(yè)實(shí)現(xiàn)數(shù)據(jù)的集中存儲(chǔ)和管理，為后續(xù)的數(shù)據(jù)分析和挖掘提供基礎(chǔ)。

此外，F(xiàn)lume還具有可靠性機(jī)制和故障轉(zhuǎn)移和恢復(fù)機(jī)制，能夠保證數(shù)據(jù)傳輸?shù)目煽啃院桶踩?。同時(shí)，F(xiàn)lume還支持客戶擴(kuò)展和自定義開(kāi)發(fā)，用戶可以根據(jù)自己的需求進(jìn)行擴(kuò)展和優(yōu)化，使其更加適合特定的應(yīng)用場(chǎng)景。

總的來(lái)說(shuō)，Apache Flume是一個(gè)功能強(qiáng)大、靈活可靠的大數(shù)據(jù)日志采集、聚合和傳輸系統(tǒng)，它在大數(shù)據(jù)處理中起到了至關(guān)重要的作用。

二、環(huán)境準(zhǔn)備

準(zhǔn)備1臺(tái)虛擬機(jī)

• Hadoop131：192.168.56.131

本例系統(tǒng)版本 CentOS-7.8，已安裝jdk1.8

關(guān)閉防火墻

systemctl stop firewalld

zookeeper、kafka 已安裝，且已啟動(dòng)

三、flume安裝配置

1、配置flume agent

1）本例演示 flume 去掉kafka數(shù)據(jù)，然后存儲(chǔ)到hdfs中
2）完整數(shù)據(jù)通道是：log文件 > flume > kafka > flume > hdfs
3）flume 安裝目錄是 /data/flume
4）kafka 、Hadoop在前面已經(jīng)安裝過(guò)

新建配置文件 /data/flume/conf/job/kafka_to_hdfs_log.conf，內(nèi)容如下：

# 定義組件
# 這里定義了Flume agent的三個(gè)主要組件：source（數(shù)據(jù)源）、channel（通道）和sink（數(shù)據(jù)接收器）。
a2.sources=r2
a2.channels=c2
a2.sinks=k2

# 配置source
# 配置數(shù)據(jù)源為Kafka，指定了Kafka的相關(guān)參數(shù)，如服務(wù)器地址、主題等。
a2.sources.r2.type = org.apache.flume.source.kafka.KafkaSource
# 每次從Kafka拉取的數(shù)據(jù)量
a2.sources.r2.batchSize=5000
# 拉取數(shù)據(jù)的間隔時(shí)間（毫秒）
a2.sources.r2.batchDurationMillis=2000
# Kafka服務(wù)器地址列表
a2.sources.r2.kafka.bootstrap.servers = hadoop131:9092,hadoop132:9092,hadoop133:9092
# 從Kafka的哪個(gè)主題拉取數(shù)據(jù)
a2.sources.r2.kafka.topics=topic_log
# 注釋掉的部分是關(guān)于攔截器的配置，攔截器可以用來(lái)對(duì)數(shù)據(jù)源的數(shù)據(jù)進(jìn)行預(yù)處理。
#a2.sources.r2.interceptors=i2
#a2.sources.r2.interceptors.i2.type = com.my.flume.interceptor.TimestampInterceptor

# 配置channel
# 配置通道為文件通道，指定了通道的相關(guān)參數(shù)，如檢查點(diǎn)目錄、數(shù)據(jù)目錄等。
a2.channels.c2.type = file
# 檢查點(diǎn)目錄，用于存儲(chǔ)通道的狀態(tài)信息
a2.channels.c2.checkpointDir = /data/flume/checkpoint/behaviorl
# 數(shù)據(jù)目錄，用于存儲(chǔ)通道中的數(shù)據(jù)
a2.channels.c2.dataDirs = /data/flume/data/behaviorl
# 通道中文件的最大大小（字節(jié)）
a2.channels.c2.maxFileSize = 2146435071
# 通道的容量，即可以存儲(chǔ)的最大事件數(shù)
a2.channels.c2.capacity = 1000000
# 通道的keepalive時(shí)間（秒）
a2.channels.c2.keepalive = 6

# 配置sink
# 配置數(shù)據(jù)接收器為HDFS，指定了HDFS的相關(guān)參數(shù)，如文件路徑、文件前綴等。
a2.sinks.k2.type =hdfs
# HDFS上的文件路徑，使用了時(shí)間變量來(lái)動(dòng)態(tài)生成目錄
a2.sinks.k2.hdfs.path = /origin_data/user/log/topic_log/%Y-%m-%d
# HDFS上的文件前綴
a2.sinks.k2.hdfs.filePrefix=log
# 是否按照時(shí)間輪轉(zhuǎn)文件，這里設(shè)置為false，表示不按照時(shí)間輪轉(zhuǎn)
a2.sinks.k2.hdfs.round =false
# 文件輪轉(zhuǎn)的時(shí)間間隔（秒）
a2.sinks.k2.hdfs.rollInterval=10
# 文件輪轉(zhuǎn)的大小閾值（字節(jié)）
a2.sinks.k2.hdfs.rollSize=134217728
# 文件輪轉(zhuǎn)的事件數(shù)閾值，這里設(shè)置為0，表示不按照事件數(shù)輪轉(zhuǎn)
a2.sinks.k2.hdfs.rollCount=0

# 控制輸出文件類型
# 設(shè)置輸出文件的類型為壓縮流格式，并使用gzip壓縮算法。
a2.sinks.k2.hdfs.fileType = CompressedStream
a2.sinks.k2.hdfs.codeC = gzip

# 組裝
# 將數(shù)據(jù)源、通道和數(shù)據(jù)接收器組裝在一起，形成一個(gè)完整的Flume agent。
a2.sources.r2.channels=c2
a2.sinks.k2.channel=c2

這個(gè)配置文件定義了一個(gè)Flume agent，它從Kafka中讀取數(shù)據(jù)，通過(guò)文件通道進(jìn)行緩存，并最終將數(shù)據(jù)寫(xiě)入到HDFS中。在寫(xiě)入HDFS時(shí)，使用了壓縮流格式，并對(duì)輸出文件進(jìn)行了gzip壓縮。同時(shí)，還通過(guò)一些參數(shù)對(duì)文件的輪轉(zhuǎn)進(jìn)行了控制。

2、啟動(dòng)flume

1）創(chuàng)建flume啟動(dòng)腳本f2.sh

vi /usr/bin/f2.sh
# 修改文件權(quán)限
chmod 777 /usr/bin/f2.sh

2）復(fù)制如下內(nèi)容

#!/bin/bash

#1. 判斷參數(shù)個(gè)數(shù)
if [ $# -lt 1 ]
then
  echo Not Enough Arguement!
  exit;
fi

case $1 in
"start")
    #遍歷集群所有機(jī)器
    for host in hadoop131
    do
        echo --------------------  $host 日志收集 flume 啟動(dòng) --------------------
        ssh $host "nohup /data/flume/bin/flume-ng agent -n a2 -c /data/flume/conf/ -f /data/flume/conf/job/kafka_to_hdfs_log.conf >/dev/null 2>&1 &"
    done
;;
"stop")
    #遍歷集群所有機(jī)器
    for host in hadoop131
    do
        echo --------------------  $host 日志收集 flume 停止 --------------------
        ssh $host "ps -ef | grep kafka_to_hdfs_log | grep -v grep | awk '{print \$2}' |xargs -n1 kill 9"
    done
;;
*)
    echo "Input Args Error..."
;;
esac

3）通過(guò)集群腳本 f2.sh 操作

f2.sh start

flume啟動(dòng)命令說(shuō)明

以下是flume啟動(dòng)命令的常用參數(shù)：

需要注意的是，F(xiàn)lume的啟動(dòng)參數(shù)可能會(huì)因版本和具體的使用場(chǎng)景而有所不同。上表中的參數(shù)是最常用的，但并不是所有的參數(shù)都在所有版本的Flume中都可用。在實(shí)際使用時(shí)，建議查閱對(duì)應(yīng)版本的Flume官方文檔或使用flume-ng agent --help命令查看可用的參數(shù)列表。

3、驗(yàn)證日志采集通路

1）在指定的log目錄中生成日志文件

cat app.log >> /data/applog/log/app_test.log

2）打開(kāi)Hadoop查看數(shù)據(jù)，http://192.168.56.131:9870/

Hadoop在前面已經(jīng)安裝過(guò)

文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-840829.html

參考

• https://flume.apache.org/

到了這里，關(guān)于【數(shù)倉(cāng)】通過(guò)Flume+kafka采集日志數(shù)據(jù)存儲(chǔ)到Hadoop的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！