本文原始碼：GitHub || GitEE
一、Flume簡介
1、基礎描述
Flume是Cloudera提供的一個高可用的，高可靠的，分布式的海量日志采集、聚合和傳輸的系統，Flume支持在日志系統中定制各類資料發送方，用于收集資料；
特點：分布式、高可用、基于流式架構，通常用來收集、聚合、搬運不同資料源的大量日志到資料倉庫，
2、架構模型
Agent包括三個核心組成，Source、Channel、Sink，Source負責接收資料源，并兼容多種型別，Channel是資料的緩沖區，Sink處理資料輸出的方式和目的地，
Event是Flume定義的一個資料流傳輸的基本單元，將資料從源頭送至目的地，
Flume可以設定多級Agent連接的方式傳輸Event資料，從最初的source開始到最終sink傳送的目的存盤系統，如果數量過多會影響傳輸速率，并且傳輸程序中單節點故障也會影響整個傳輸通道，
Flume支持多路復用資料流到一個或多個目的地，這種模式可以將相同資料復制到多個channel中，或者將不同資料分發到不同的channel中，并且sink可以選擇傳送到不同的目的地，
Agent1理解為路由節點負責Channel的Event均衡到多個Sink組件，每個Sink組件分別連接到獨立的Agent上，實作負載均衡和錯誤恢復的功能，
Flume的使用組合方式做資料聚合，每臺服務器部署一個flume節點采集日志資料，再匯聚傳輸到存盤系統，例如HDFS、Hbase等組件，高效且穩定的解決集群資料的采集，
二、安裝程序
1、安裝包
apache-flume-1.7.0-bin.tar.gz
2、解壓命名
[root@hop01 opt]# pwd
/opt
[root@hop01 opt]# tar -zxf apache-flume-1.7.0-bin.tar.gz
[root@hop01 opt]# mv apache-flume-1.7.0-bin flume1.7
3、組態檔
配置路徑：/opt/flume1.7/conf
mv flume-env.sh.template flume-env.sh
4、修改配置
添加JDK依賴
vim flume-env.sh
export JAVA_HOME=/opt/jdk1.8
5、環境測驗
安裝netcat工具
sudo yum install -y nc
創建任務配置
[root@hop01 flume1.7]# cd job/
[root@hop01 job]# vim flume-netcat-test01.conf
添加基礎任務配置
注意：a1表示agent名稱，
# this agent
a1.sources = sr1
a1.sinks = sk1
a1.channels = sc1
# the source
a1.sources.sr1.type = netcat
a1.sources.sr1.bind = localhost
a1.sources.sr1.port = 55555
# the sink
a1.sinks.sk1.type = logger
# events in memory
a1.channels.sc1.type = memory
a1.channels.sc1.capacity = 1000
a1.channels.sc1.transactionCapacity = 100
# Bind the source and sink
a1.sources.sr1.channels = sc1
a1.sinks.sk1.channel = sc1
開啟flume監聽埠
/opt/flume1.7/bin/flume-ng agent --conf /opt/flume1.7/conf/ --name a1 --conf-file /opt/flume1.7/job/flume-netcat-test01.conf -Dflume.root.logger=INFO,console
使用netcat工具向55555埠發送資料
[root@hop01 ~]# nc localhost 55555
hello,flume
查看flume控制面
三、應用案例
1、案例描述
基于flume在各個集群服務進行資料采集，然后資料傳到kafka服務，再考慮資料的消費策略，
采集：基于flume組件的便捷采集能力，如果直接使用kafka會產生大量的埋點動作不好維護，
消費：基于kafka容器的資料臨時存盤能力，避免系統高度活躍期間采集資料過大沖垮資料采集通道，并且可以基于kafka做資料隔離并針對化處理，
2、創建kafka配置
[root@hop01 job]# pwd
/opt/flume1.7/job
[root@hop01 job]# vim kafka-flume-test01.conf
3、修改sink配置
# the sink
a1.sinks.sk1.type = org.apache.flume.sink.kafka.KafkaSink
# topic
a1.sinks.sk1.topic = kafkatest
# broker地址、埠號
a1.sinks.sk1.kafka.bootstrap.servers = hop01:9092
# 序列化方式
a1.sinks.sk1.serializer.class = kafka.serializer.StringEncoder
4、創建kafka的Topic
上述組態檔中名稱：kafkatest，下面執行創建命令之后查看topic資訊，
[root@hop01 bin]# pwd
/opt/kafka2.11
[root@hop01 kafka2.11]# bin/kafka-topics.sh --create --zookeeper hop01:2181 --replication-factor 1 --partitions 1 --topic kafkatest
[root@hop01 kafka2.11]# bin/kafka-topics.sh --describe --zookeeper hop01:2181 --topic kafkatest
5、啟動Kakfa消費
[root@hop01 kafka2.11]# bin/kafka-console-consumer.sh --bootstrap-server hop01:2181 --topic kafkatest --from-beginning
這里指定topic是kafkatest，
6、啟動flume配置
/opt/flume1.7/bin/flume-ng agent --conf /opt/flume1.7/conf/ --name a1 --conf-file /opt/flume1.7/job/kafka-flume-test01.conf -Dflume.root.logger=INFO,console
四、源代碼地址
GitHub·地址
https://github.com/cicadasmile/big-data-parent
GitEE·地址
https://gitee.com/cicadasmile/big-data-parent
閱讀標簽
【Java基礎】【設計模式】【結構與演算法】【Linux系統】【資料庫】
【分布式架構】【微服務】【大資料組件】【SpringBoot進階】【Spring&Boot基礎】
【資料分析】【技術導圖】【 職場】
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/266631.html
標籤：其他