目錄

1. Flume介紹

Flume是Cloudera提供的一個分散式、可靠、和高可用的海量日誌採集、聚合和傳輸的日誌收集系統，支援在日誌系統中定製各類資料傳送方，用於收集資料;同時，Flume提供對資料進行簡單處理，並寫到各種資料接受方(可定製)的能力。

1.1 Flume資料來源以及輸出方式

Flume提供了從console(控制檯)、RPC(Thrift-RPC)、text(檔案)、tail(UNIX tail)、syslog(syslog日誌系統，支援TCP和UDP等2種模式)，exec(命令執行)等資料來源上收集資料的能力,在我們的系統中目前使用exec方式進行日誌採集。

Flume的資料接受方，可以是console(控制檯)、text(檔案)、dfs(HDFS檔案)、RPC(Thrift-RPC)和syslogTCP(TCP syslog日誌系統)等。本測試研究中由kafka來接收資料。

1.2 Flume的核心概念

1.  Agent：使用JVM執行Flume。每臺機器執行一個agent，但是可以在一個agent中包含多個sources和sinks。

2.  Client：生產資料，執行在一個獨立的執行緒。

3.  Source：從Client收集資料，傳遞給Channel。

4.  Sink：從Channel收集資料，執行在一個獨立執行緒。

5.  Channel

6.  Events ：可以是日誌記錄、 avro物件等。

1.3 結構

Flume以agent為最小的獨立執行單位。一個agent就是一個JVM。單agent由Source、Sink和Channel三大元件構成，如下圖：

Flume提供了大量內建的Source、Channel和Sink型別。不同型別的Source,Channel和Sink可以自由組合。組合方式基於使用者設定的配置檔案，非常靈活。比如：Channel可以把事件暫存在記憶體裡，也可以持久化到本地硬碟上。Sink可以把日誌寫入HDFS, HBase，甚至是另外一個Source等等。Flume支援使用者建立多級流，也就是說，多個agent可以協同工作，並且支援Fan-in、Fan-out、ContextualRouting、Backup Routes。如下圖所示:

1.4 安裝測試

解壓apache-flume-1.6.0-bin.tar.gz：tar –zxvf apache-flume-1.6.0-bin.tar.gz

cp conf/flume-conf.properties.template conf/exec.conf

cp conf/flume-env.sh.template conf/flume-env.sh    配置JAVA_HOME

exec.conf配置如下:

a2.sources = r2

a2.sinks = k2

a2.channels = c2

# Describe/configure the source

a2.sources.r2.type = exec

a2.sources.r2.channels = c2

a2.sources.r2.command=tail -n +0 -F /usr/local/hadoop/flume/test.log

# Describe the sink

a2.sinks.k2.type = logger

# Use a channel which buffers events in memory

a2.channels.c2.type = memory

a2.channels.c2.capacity = 1000

a2.channels.c2.transactionCapacity = 100

# Bind the source and sink to the channel

a2.sources.r2.channels = c2

a2.sinks.k2.channel = c2

驗證安裝：flume-ng version

1.5 啟動flume

flume-ng agent --conf ./flume/conf/ -f ./flume/conf/exec.conf-Dflume.root.logger=DEBUG,console -n a2

傳送資料和flume接收資料：

2.Kafka介紹

2.1 產生背景

Kafka 是分散式釋出-訂閱訊息系統。它最初由 LinkedIn 公司開發，使用 Scala語言編寫,之後成為 Apache 專案的一部分。Kafka是一個分散式的，可劃分的，多訂閱者,冗餘備份的永續性的日誌服務。它主要用於處理活躍的流式資料。

在大資料系統中，常常會碰到一個問題，整個大資料是由各個子系統組成，資料需要在各個子系統中高效能，低延遲的不停流轉。傳統的企業訊息系統並不是非常適合大規模的資料處理。為了已在同時搞定線上應用（訊息）和離線應用（資料檔案，日誌）Kafka 就出現了。Kafka 可以起到兩個作用：

降低系統組網複雜度

降低程式設計複雜度，各個子系統不在是相互協商介面，各個子系統類似插口插在插座上，Kafka 承擔高速資料匯流排的作用。

2.2 部署結構

2.3 叢集架構

2.4 基本概念

Topic：特指 Kafka 處理的訊息源（feeds of messages）的不同分類。

Partition：Topic 物理上的分組，一個topic 可以分為多個 partition，每個 partition 是一個有序的佇列。partition 中的每條訊息都會被分配一個有序的id（offset）。

Message：訊息，是通訊的基本單位，每個 producer 可以向一個 topic（主題）釋出一些訊息。

Producers：訊息和資料生產者，向 Kafka的一個 topic 釋出訊息的過程叫做 producers。

Consumers：訊息和資料消費者，訂閱 topics 並處理其釋出的訊息的過程叫做 consumers。

Broker：快取代理，Kafa 叢集中的一臺或多臺伺服器統稱為broker。

2.5 安裝測試

解壓Kafka: tar -xzf kafka_2.10-0.8.1.1.tgz

啟動ZK bin/zookeeper-server-start.shconfig/zookeeper.properties

啟動服務bin/kafka-server-start.sh config/server.properties>/dev/null 2>&1&

建立主題 bin/kafka-topics.sh --create --zookeeperlocalhost:2181 --replication-factor 1 --partitions 1 --topic test

檢視主題 bin/kafka-topics.sh --list --zookeeperlocalhost:2181

檢視主題詳情 bin/kafka-topics.sh--describe --zookeeper localhost:2181 --topic test

刪除主題 bin/kafka-run-class.shkafka.admin.DeleteTopicCommand --topic test --zookeeper localhost:2181

建立生產者bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

建立消費者bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test--from-beginning

3.Flume和Kafka整合

3.1 兩者整合優勢

Flume更傾向於資料傳輸本身，Kakfa是典型的訊息中介軟體用於解耦生產者消費者。

具體架構上，Agent並沒把資料直接傳送到Kafka，在Kafka前面有層由Flume構成的forward。這樣做有兩個原因：

Kafka的API對非JVM系的語言支援很不友好，forward對外提供更加通用的HTTP介面。

forward層可以做路由、Kafka topic和Kafkapartition key等邏輯，進一步減少Agent端的邏輯。

資料有資料來源到flume再到Kafka時，資料一方面可以同步到HDFS做離線計算，另一方面可以做實時計算。本文實時計算採用SparkStreaming做測試。

3.2 Flume和Kafka整合安裝

提取外掛中的flume-conf.properties檔案：修改如下:flume源採用exec

producer.sources.s.type = exec

producer.sources.s.command=tail -f -n+1/usr/local/Hadoop/flume/test.log

producer.sources.s.channels = c

修改producer代理的topic為test

將配置放到flume/cong/producer.conf中

複製外掛包中的jar包到flume/lib中：刪除掉版本不同的相同jar包，這裡需要刪除scala-compiler-z.9.2.jar包，否則flume啟動會出現問題。

複製kafka/libs中的jar包到flume/lib中。

完整producer.conf:

producer.conf:

#agentsection 

producer.sources= s 

producer.channels= c 

producer.sinks= r 

#sourcesection 

producer.sources.s.type= exec

#producer.sources.s.spoolDir= /usr/local/hadoop/flume/logs

#producer.sources.s.fileHeader= true

producer.sources.s.command= tail -f -n+1 /usr/local/hadoop/flume/aaa.log

producer.sources.s.channels= c 

# Eachsink's type must be defined 

producer.sinks.r.type= org.apache.flume.plugins.KafkaSink 

producer.sinks.r.metadata.broker.list=localhost:9092 

producer.sinks.r.partition.key=0 

producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition 

producer.sinks.r.serializer.class=kafka.serializer.StringEncoder 

producer.sinks.r.request.required.acks=0 

producer.sinks.r.max.message.size=1000000 

producer.sinks.r.producer.type=sync 

producer.sinks.r.custom.encoding=UTF-8 

producer.sinks.r.custom.topic.name=test 

#Specifythe channel the sink should use 

producer.sinks.r.channel= c 

# Eachchannel's type is defined. 

producer.channels.c.type= memory 

producer.channels.c.capacity= 1000

producer.channels.c.transactionCapacity= 100

3.3 啟動kafka flume相關服務

啟動ZKbin/zookeeper-server-start.sh config/zookeeper.properties

啟動Kafka服務 bin/kafka-server-start.sh config/server.properties

建立消費者bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test--from-beginning

啟動flume

flume-ng agent --conf./flume/conf/ -f ./flume/conf/producer.conf -Dflume.root.logger=DEBUG,console-n producer

向flume傳送資料：

Kafka消費者資料：

3.4 Kafka和SparkStreaming整合

核心程式碼：

完整程式碼路徑：

spark-1.4.0\examples\src\main\java\org\apache\spark\examples\streaming

執行引數：

傳送資料：

由於flume採用exec資料來源的方式，因此flume會監聽配置的相應的檔案： tail -f -n+1 /usr/local/Hadoop/flume/aaa.log

當向該檔案追加檔案時，flume就會獲取追加的資料：

writetoflume.py

flume將獲取的增量資料由sink傳送給kafka,以下是kafka comsumer消費的資料

執行結果：

SparkStreaming訂閱kafka的test主題的資料，將訂閱的資料進行單詞計數處理。