2. 程式人生 > >flume-1.7.0 簡單使用

flume-1.7.0 簡單使用

阿新 • • 發佈：2019-01-05

在上一篇中，我們安裝了 flume-ng，這一篇我們就來簡單使用一下。

這裡寫圖片描述

官網上是這麼介紹的，我們需要指定一個配置檔案，需要定義一個 agent 的名稱，然後我們就可以使用 flume-ng 命令來啟動了。

1 編寫配置檔案

我們先拿官網上的例子來跑一下看看，就使用 example.conf 檔案：

[[email protected] conf]# pwd
/usr/hadoop/flume-1.7.0-bin/conf
[[email protected] conf]# vi example.conf 

# example.conf: A single-node Flume configuration
 


# Name the components on this agent
# 定義一個 agent 的元素

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 配置 source

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
# 配置 sink

a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
 

# 定義 channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
# 用 channel 連線起來 source 和 sink

a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

這裡我簡單寫了一下注釋，這可不是翻譯過來的昂，大家看的時候別被我誤導了。具體概念大家還是看官網咖。
其中有四個概念：
這裡寫圖片描述


這是官網上給的示意圖，我們可以這樣理解：
WebServer 想給 HDFS 送點東西 (source) ，於是就找了 flume 這個 agent(代理) ，然後 WebServer 把 source 給了 agent , agent 拿到之後，用它自己的手段(channel)，可能是”物流“，然後到了離 HDFS 最近的”快遞分揀點“(sink)，把東西給了 HDFS。
這張圖只是示意圖，別不是對於 example.conf 的解釋，希望大家別想多了。

1.1 source

別聽我胡扯，我只是為了方便理解，自己意淫的。
那麼，對於這個 example.conf 的配置檔案，我們定義了一個叫 a1 的 agent，然後 source 源配置的是 netcat 型別，對於 netcat source 需要配置的內容官網上寫的很清楚：
這裡寫圖片描述

黑色加粗的幾項是必須配置的，對於其他幾項是可選項。

1.2 channel

接下來是 channel，也就是我們要選那一種”物流“，這裡我們用的是 memory，我們需要配置的是：
這裡寫圖片描述

1.3 sink

我們的”物流分揀點“，sink 我們配置的是 logger，需要配置幾項是：
這裡寫圖片描述

可能你已經被我誤導了，對於他們真正的解釋還是看官網，獲取你對它們的理解，把這些東西轉換成為你能夠理解的東西就行了。

2 啟動 agent

2.1 啟動

上面我們已經說過了，使用 flume-ng 命令啟動。
這裡寫圖片描述

具體引數看上圖。

[[email protected] conf]# pwd
/usr/hadoop/flume-1.7.0-bin/conf

[[email protected] conf]# flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=INFO,console

Info: Including Hadoop libraries found via (/usr/hadoop/hadoop-2.6.4/bin/hadoop) for HDFS access
Info: Including HBASE libraries found via (/usr/hadoop/hbase-1.2.3/bin/hbase) for HBASE access
...

16/11/18 19:34:27 INFO node.Application: Starting Channel c1
16/11/18 19:34:27 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: CHANNEL, name: c1: Successfully registered new MBean.
16/11/18 19:34:27 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started
16/11/18 19:34:27 INFO node.Application: Starting Sink k1
16/11/18 19:34:27 INFO node.Application: Starting Source r1
16/11/18 19:34:27 INFO source.NetcatSource: Source starting
16/11/18 19:34:27 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]

想必大家也都注意到了，我所在的目錄是 conf，因為我們需要指定一個配置檔案，需要它的絕對路徑。然後看到上面的情況就表示我們的名為 a1 的 agent 啟動成功了。由於我們配置的 sink 是 logger ，並且指定了一些列引數，把內容輸出到我們的控制檯。

2.2 傳送訊息

接下來我們就可以再開一個終端，在這個終端上執行下面一系列命令：
這裡寫圖片描述

可能有些朋友的Linux上會提示 command not found，只要安裝一下 telnet 就可以了。這裡我們可以看到一個”Connection refused“，是因為我們的 /etc/hosts 檔案中 localhost 對應的 ”::1…“這樣形式的 ip 沒有識別成功，它嘗試了”127.0.0.1“就連上了。具體的還是要看關於網路這塊兒的內容，我就不細講了。

這個時候，我們可以在這兒輸入一些東西：
這裡寫圖片描述

然後我們回到之前執行著 agent 的那個終端：
這裡寫圖片描述

會看到多了這樣一行內容，可能有時候這一行顯示的內容比我們輸入的內容要少，並不是沒有接收到，是因為超過了它能顯示的長度，給省略掉了。

這個時候應該就有所體會了，flume 是”很多個形容詞“的
這裡寫圖片描述

日誌採集系統。

我們再來寫幾個例項，來體會一下。

3. avro ⇒ hdfs

我們的 source 是 avro，sink 是 hdfs，（我這種說法嚴格來說是不正規的，但是我不知道怎麼說你們能夠理解，暫時就先這樣認為）。
那麼，先找找看 avro source 需要配置的是什麼。

3.1 配置 avro source

這裡寫圖片描述
我們需要配置有四項，並且下面也給出了示例。

[[email protected] conf]# vi AvroHDFS.conf
a1.sources = r1
a1.channels = c1

a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = master
a1.sources.r1.port = 4141

3.2 配置 channel

這次我們還是使用 memory：
這裡寫圖片描述

我們在 AvroHDFS.conf 檔案中追加：

[[email protected] conf]# vi AvroHDFS.conf
a1.sources = r1
a1.channels = c1

a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = master
a1.sources.r1.port = 4141

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000

3.3 配置 sink

我們是向 HDFS 上導資料，所以我們使用 hdfs sink：

這裡寫圖片描述
這部分圖片太長了，放上來大家也看不清，於是我就只截了示例，大家還是去官網上看看怎麼配置吧。

我們接著在 AvroHDFS.conf 檔案中追加：

[[email protected] conf]# vi AvroHDFS.conf
a1.sinks = k1
a1.sources = r1
a1.channels = c1

a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = master
a1.sources.r1.port = 4141

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000

a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /fromflume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events.
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true

這樣我們就定義好了一個從 avro 到 hdfs 的配置檔案，我們可以啟動 agent 了。

3.4 啟動 agent

這次我們需要先執行起來 hadoop 叢集，不然是會失敗的。

[[email protected] conf]# flume-ng agent --conf conf --conf-file AvroHDFS.conf --name a1 -Dflume.root.logger=INFO,console

Info: Including Hadoop libraries found via (/usr/hadoop/hadoop-2.6.4/bin/hadoop) for HDFS access
Info: Including HBASE libraries found via (/usr/hadoop/hbase-1.2.3/bin/hbase) for HBASE access
Info: Including Hive libraries found via (/usr/hadoop/apache-hive-2.1.0-bin) for Hive access
...

16/11/18 20:46:13 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
16/11/18 20:46:13 INFO source.AvroSource: Avro source r1 started.

這樣就表示我們的 agent 啟動成功了。不過接下來我們需要使用API 了，java程式如下：

import java.nio.charset.Charset;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

public class FlumeDemo {
    private String hostname;
    private int port;
    private RpcClient client;

    public FlumeDemo(String hostname,int port) {
        this.hostname = hostname;
        this.port = port;
        this.client = RpcClientFactory.getDefaultInstance(hostname, port);
    }

    public void sendMessage(String data){
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            e.printStackTrace();
        }
    }

    public void cleanUp(){
        client.close();
    }

    public static void main(String[] args) {
        FlumeDemo rpcClient = new FlumeDemo("master", 4141);

        String data = "testing ";

        for(int i=0;i<10;i++){
            rpcClient.sendMessage(data + i);
        }

        rpcClient.cleanUp();
    }
}

然後，執行我們的 java 程式，這個時候，觀察我們的 agent 是什麼狀況：

...
16/11/18 20:46:13 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
16/11/18 20:46:13 INFO source.AvroSource: Avro source r1 started.

16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 => /192.168.38.129:4141] OPEN
16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 => /192.168.38.129:4141] BOUND: /192.168.38.129:4141
16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 => /192.168.38.129:4141] CONNECTED: /192.168.38.1:64375
16/11/18 20:46:23 INFO hdfs.HDFSSequenceFile: writeFormat = Writable, UseRawLocalFileSystem = false
16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 :> /192.168.38.129:4141] DISCONNECTED
16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 :> /192.168.38.129:4141] UNBOUND
16/11/18 20:46:23 INFO ipc.NettyServer: [id: 0x33dfe52f, /192.168.38.1:64375 :> /192.168.38.129:4141] CLOSED
16/11/18 20:46:23 INFO ipc.NettyServer: Connection to /192.168.38.1:64375 disconnected.

16/11/18 20:46:24 INFO hdfs.BucketWriter: Creating /fromflume/events/16-11-18/2040/00/events..1479473183696.tmp
...

這個時候，我們可以看到，已經在往 HDFS 上寫資料了，我們可以通過WebUI(通過瀏覽器訪問：http://master:50070 )開看看 hdfs 上是不是多了 /fromflume/… 的資料夾
這裡寫圖片描述

果然，我們的資料夾已經建立成功了，我們可以一級一級的進去，會看到：
這裡寫圖片描述

檔案已經成功寫入，我們檢視的時候肯定是各種亂碼的…但是我們是已經成功寫入了。

4 avro ⇒ kafka

結合我昨天寫的，我們來寫一下 flume 和 kafka 結合的一個例子。兩者有點相似，都是”代理/中介“。

有了前面兩個練習，那這個我就不寫的那麼詳細了。

這裡寫圖片描述
但是，對於這個 kafka sink 還真是有好多需要說的：

4.1 sinks.type

這個必須設定成”org.apache.flume.sink.kafka.KafkaSink“，你們說這個也是奇怪，這一項沒有預設值，要設定的時候還 ” Must be set to org.apache.flume.sink.kafka.KafkaSink“。

4.2 kafka.topic

關於這一項，那裡也有提到，大概意思就是，當 topic 在 event 的 header 中時，kafka 的 broker 中原來跟這個 topic 同名的就會被覆寫。……我不想翻譯了，我有點繞暈了。等我想好了再來補充這一部分。
這一塊暫時還用不到，現在不理解也不打緊。先記住有這麼一茬。

好了，我們來寫一下 sink ：

[[email protected] conf]# vi AvroKafka.conf
      1 a1.sources = r1
      2 a1.sinks = k1
      3 a1.channels = c1
      4 
      5 a1.sources.r1.type = avro
      6 a1.sources.r1.bind = master
      7 a1.sources.r1.port = 44444
      8 
      9 
     10 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
     11 a1.sinks.k1.kafka.bootstrap.servers = localhost:9092,localhost:9093
     12 a1.sinks.k1.kafka.topic = fromFlume 
     13 
     14 a1.channels.c1.type = memory
     15 a1.channels.c1.capacity = 1000
     16 a1.channels.c1.transactionCapacity = 100
     17 
     18 a1.sources.r1.channels = c1
     19 a1.sinks.k1.channel = c1

要 flume 和 kafka 結合，這次我們是讓 kafka 消費 flume 發出去的資料。
那麼，我們需要啟動 kafka 的服務，並且建立一個消費者，為了演示，這次我們就在 master 上啟動兩個 broker 好了。

[[email protected] config]# pwd
/usr/hadoop/kafka_2.11-0.10.1.0/config
[[email protected] config]# kafka-server-start.sh server.properties &
[1] 10693
...
[[email protected] config]# kafka-server-start.sh server1.properties &
[2] 10962
...
[[email protected] config]# jps
11233 Jps
2593 ResourceManager
10962 Kafka
2692 NodeManager
10693 Kafka
3034 QuorumPeerMain
2171 NameNode
2269 DataNode
2446 SecondaryNameNode
[[email protected] config]# 

[[email protected] config]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic fromFlume --from-beginning

這個時候，我們就可以啟動我們的 flume agent 了：

[[email protected] conf]# flume-ng agent --conf conf --conf-file AvroKafka.conf --name a1 -Dflume.root.logger=INFO,console

...

不知道大家有沒有暈掉，反正我當初學的時候是暈了好一陣兒呢。^8^

接下來，我們還使用之前的 java 程式：
不過這時候的埠號需要修改一下，記住，這個埠號需要跟我們的 *.conf 檔案中配置的埠號一致。


import java.nio.charset.Charset;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

public class FlumeDemo {
    private String hostname;
    private int port;
    private RpcClient client;

    public FlumeDemo(String hostname,int port) {
        this.hostname = hostname;
        this.port = port;
        this.client = RpcClientFactory.getDefaultInstance(hostname, port);
    }

    public void sendMessage(String data){
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            e.printStackTrace();
        }
    }

    public void cleanUp(){
        client.close();
    }

    public static void main(String[] args) {
        FlumeDemo rpcClient = new FlumeDemo("master", 44444);

        String data = "Hello World! ";

        for(int i=0;i<10;i++){
            rpcClient.sendMessage(data + i);
        }

        rpcClient.cleanUp();
    }
}

執行程式，然後，我們去到啟動 kafka 消費者的那個終端上，我們會看到：

[2016-11-18 21:46:54,915] INFO [Group Metadata Manager on Broker 0]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2016-11-18 21:47:08,073] INFO [Group Metadata Manager on Broker 1]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
Hello World! 0
Hello World! 1
Hello World! 2
Hello World! 3
Hello World! 4
Hello World! 5
Hello World! 6
Hello World! 7
Hello World! 8
Hello World! 9

不知道大家想到了什麼，反正我是一度這跟輸出重定向似的。我都說了，你們別聽我胡扯^v^。
這些還都只是簡單的小例項，大家可以好好看看文件，看看 flume 都是有什麼 sources、channels、sinks，自己整著玩玩兒，玩著玩著就上手了。

相關推薦

flume-1.7.0 簡單使用

在上一篇中，我們安裝了 flume-ng，這一篇我們就來簡單使用一下。 官網上是這麼介紹的，我們需要指定一個配置檔案，需要定義一個 agent 的名稱，然後我們就可以使用 flume-ng 命令來啟動了。 1 編寫配置檔案 我們先拿官網上的例子來跑

Flume NG高可用叢集搭建詳解（基於flume-1.7.0）

1、Flume NG簡述 Flume NG是一個分散式，高可用，可靠的系統，它能將不同的海量資料收集，移動並存儲到一個數據儲存系統中。輕量，配置簡單，適用於各種日誌收集，並支援 Failover和負載均衡。並且它擁有非常豐富的元件。Flume NG採用的是三層架構：Agent層，Collecto

Flume 1.7.0安裝與例項

Flume安裝 系統要求： 需安裝JDK 1.7及以上版本 2、解壓 $ cp ~/Downloads/apache-flume-1.7.0-bin.tar.gz ~ $ cd $ tar -zxvf apache-flume-1.7.0-bi

flume 1.7.0－taildirSource 支援 windows系統

Flume-ng 1.7.0 中增加了TaildirSource，可以監控目錄中檔案的變化自動讀取檔案內容。 不過實際應用時發現幾個問題： 1，不支援windows系統。 2，windows下會影響 log4j 日誌檔案的切分，會使log4j日誌不切分一直

Rust 1.7.0 語法基礎 sep_token 和 non_special_token

++ pri ava type mod ont key 代碼 1.7 一、分隔符 sep_token 指的是分隔符, 是除了 * 和 + 之外的不論什麽符號，通常情況下是使用 ， 逗號。 比如： 宏的多個參數分隔,以下代碼中的逗號就是 sep_

Rust 1.7.0 匹配器 match 的簡介和使用

let 滿足 選擇 多個 efault msg i++ pretty article 使用過正則表達式的人應該都知道 matcher ,通過 matcher 匹配器運算正則表達式，完畢一系列的匹配規則。 在Rust 中 沒有 switch 語句。mat

開發常見錯誤之 ：Missing artifact com.sun:tools:jar 1.7.0

problems code iss file http 1.7 tin osi 改名 Missing artifact com.sun:tools:jar 1.7.0 解決辦法一： 手動配置pom.xml，添加一個dependency如下： <dependency&

flume 1.7 的配置

zxvf 解決 tex clas 文件 env home 收集 bin Apache Flume是一個分布式的、可靠的、高效的日誌數據收集組件；我們通常使用Flume將分散在集群中多個Servers的log文件，匯集到中央式的數據平臺中，以解決“從離散的日誌文件中查看、統

Navicat Premium 12.1.7.0安裝與啟用

本文介紹Navicat Premium 12.1.7.0的安裝、啟用與基本使用。 博主所提供的啟用檔案理論支援Navicat Premium 12.0.x系列和Navicat Premium 12.1.x系列的註冊機。由於本文一直在更新，Navicat Premium 12.0.x系列

Windows10離線安裝Anaconda3-4.2.0-Windows-x86_64.exe（對應python3.5）和tensorflow_gpu-1.7.0-cp35-cp35m-win_amd

Windows10離線安裝Anaconda3-4.2.0-Windows-x86_64.exe（對應python3.5）和tensorflow_gpu-1.7.0-cp35-cp35m-win_amd64.whl（對應GPU版本的tensorflow，35表示著對應python3.5） 安裝這個

ros_rplidar 1.7.0大升級

　　ROS作為機器人軟體平臺，具有優良的開源基因，鬆耦合設計方法，可擴充套件的軟體結構，在研究機器人行走等方面起著重要的作用。Lidar作為機器人定位導航的核心感測器，在機器人自主定位、建圖及後續導航方面也扮演著重要角色。兩者的結合使用，使得機器人自主定位導航效果更佳，使用門檻更低。 　　

Apache Flink 1.7.0 釋出，支援 Scala 2.12

Apache Flink 1.7.0 已釋出，Apache Flink 是一個開源的流處理框架，應用於分散式、高效能、始終可用的、準確的資料流應用程式。 新特性和改進： 支援 Scala 2.12 State Evolution：現在能夠更靈活地調整長時間執行的應用的使用者狀態模式，同時保持與先

lapis 1.7.0 更好的openresty 版本相容以及安全資料庫支援

lapis 1.7.0 今年4月2號就釋出了，一直沒有注意，今天看到changelog就簡單的進行了一個 測試（主要是與openresty版本的測試，新變更後邊會有） 使用docker-compose 執行 環境準備 docker-compose 檔案 version: "3

Spark中分散式使用HanLP（1.7.0)分詞d

Spark中分散式使用HanLP（1.7.0)分詞 HanLP分詞（https://github.com/hankcs/HanLP），如README中所說，如果沒有特殊需求，可以通過maven配置，如果要新增自定義詞典，需要下載“依賴jar包和使用者字典". 本人一些經驗:

Flume 1.8.0 開發者指南-個人翻譯版

（為了方便學習利用週末翻譯出來的，現在拿出來和大家免費分享，歡迎轉載，但請尊重我個人的勞動成果，請標明來源） （歡迎在評論中交流。如果覺得我翻譯的還可以歡迎讚美和鼓勵，如果發現了翻譯不當之處也歡迎您無須客氣的指出。謝謝！） 介紹 概覽 Apache Flume 是一個

Apache Flume 1.7深入淺出，快速入門

汪文君Flume深入淺出實戰視訊-01-ApacheFlume的介紹.wmv 汪文君Flume深入淺出實戰視訊-02-環境搭建，快速入門，快速啟動一個Agent.wmv 汪文君Flume深入淺出實戰視訊-03-Apache Flume的配置，Avro Agent的詳細介紹.

Kubeadm offline installer 升級到 1.7.0 版本遇到的坑_Kubernetes中文社群

本來做這玩意的初衷就是，Kubeadm 和 Kubernetes 是一家人，升級比較方便跟得住。未曾想第一次大版本升級，就遇到了個不大不小的坑，導致安裝無法完成。這個 Issue 會在 1.7.1 修補，下面介紹一下曲線救國的安裝方式。 這一問題的似乎是 kubeadm 的更新破壞了 TLS

Performance Improvements in Mesos 1.7.0

Performance Improvements in Mesos 1.7.0Scalability and performance are key features for Mesos. Some users of Mesos already run production clusters that con

Rust 1.7.0的macro巨集－語法分析和使用舉例

macro 巨集的概念在很多語言中都有。 通常情況下，巨集的機制是在預編譯階段對已經定義的巨集進行替換或者 expanded 展開 ，即：把巨集按照名稱替換成巨集的內容。 Rust 中的巨集，與眾不同！ Rust中的巨集，也是在預編譯階段進行處理。巨集不僅

Apache Flume 1.6.0 使用者指南

$ telnet localhost 44444 Trying 127.0.0.1... Connected to localhost.localdomain (127.0.0.1). Escape character is '^]'. Hello world! <ENTER> OK 在