:-: **常用的Source類型** | 類型 | 描述 | | --- | --- | | netcat source（TCP） | 它打開指定的端口并偵聽數據。期望提供的數據是換行分隔的文本。每行文本被轉換成一個 Flume 事件并通過連接的通道發送。 | | exec source | 執行Linux指令，并消費指令返回的結果。 | | spooling directory source | 該 Source 將監視指定目錄的新文件，一旦有新文件產生立即解析它們。將給定的文件完全讀入Channel后，默認情況下通過重命名該文件來表示完成，或者可以刪除該文件，或者使用 trackerDir 跟蹤已處理的文件。 | | http source | 用于接收HTTP的Get和Post請求。 | | avro source | 監聽 Avro 端口并從外部 Avro 客戶端流接收事件。當與另一個(前一跳)Flume Agent 上的 Avro Sink 一起使用時，它可以創建分層收集拓撲。 | | kafka source | 它讀取來自Kafka主題的消息。 | | taildir source|可以同時監控一個或多個文件, 并帶有偏移量存儲文件來記錄上次讀到的位置, 下次可以接著讀。<br/> 注意：taildir source 目前不能運行在 windows 系統上。 | **如何使用這些不同的Source：** （1）查看官方文檔 https://flume.apache.org/releases/content/1.8.0/FlumeUserGuide.html 檢查Source都有哪些配置。如下為 exec source的配置：  （2）按照【Flume使用過程】這一節將Source的相關配置替換。<br/> 下面就上表列出的Source分別做一個使用例子。<br/> [TOC] # 1. exec source 舉例 （1）創建一個`exec_source.conf`文件，編寫Agent、Source、Channel、Sink的配置信息。 ```xml ############## 1. Agent初始化 ############ # agent為Agent的名字，可以隨便命名 # s1、c1、sk1是在該agent下的Source、Channel、Sink，也可以隨便命名 agent.sources = s1 agent.channels = c1 agent.sinks = sk1 ############## 2. Source配置 ############# # 設置Source的類型、Linux命令、通道 agent.sources.s1.type = exec # tail命令可以實時檢測一個文件的內容是否發生了更改 agent.sources.s1.command = tail -f /opt/install/flume/myconf/learn.log agent.sources.s1.channels = c1 ############## 3. Sink配置 ############# # 設置Sink類型為logger模式 agent.sinks.sk1.type = logger # Sink從c1通道獲取數據 agent.sinks.sk1.channel = c1 ############ 4. Channel配置 ############ # 設置Channel為內存模式，容量1000，傳輸參數100 agent.channels.c1.type = memory ``` （2）啟動Agent服務 ```shell -- 切換到flume的根目錄下執行下面的語句。當然如果你已經配置了環境變量則就不需要了 bin/flume-ng agent -c conf -f myconf/exec_source.conf --name agent -Dflume.root.logger=INFO,console ``` 可以看到打印出了learn.log文件的內容。  （3）當你改變learn.log文件的內容時，會被監測到。  <br/> # 2. spooling directory source 舉例 （1）創建一個`spooldir_source.conf`文件，編寫Agent、Source、Channel、Sink的配置信息。 ```xml ############## 1. Agent初始化 ############ # agent為Agent的名字，可以隨便命名 # s1、c1、sk1是在該agent下的Source、Channel、Sink，也可以隨便命名 agent.sources = s1 agent.channels = c1 agent.sinks = sk1 ############## 2. Source配置 ############# # 設置Source的類型、需要監控的目錄、通道 agent.sources.s1.type = spooldir agent.sources.s1.spoolDir = /opt/install/flume/myconf/ agent.sources.s1.channels = c1 ############## 3. Sink配置 ############# # 設置Sink類型為logger模式 agent.sinks.sk1.type = logger # Sink從c1通道獲取數據 agent.sinks.sk1.channel = c1 ############ 4. Channel配置 ############ # 設置Channel為內存模式 agent.channels.c1.type = memory ``` （2）啟動Agent服務 ```shell -- 切換到flume的根目錄下執行下面的語句。當然如果你已經配置了環境變量則就不需要了 bin/flume-ng agent -c conf -f myconf/spooldir_source.conf --name agent -Dflume.root.logger=INFO,console ``` （3）向/opt/install/flume/myconf新建一個test.log文件，查看Flume的輸出如下：  （4）查看/opt/install/flume/myconf下的所有文件都被新增了`.COMPLETED`后綴  <br/> # 3. http source 舉例 （1）創建一個`http_source.conf`文件，編寫Agent、Source、Channel、Sink的配置信息。 ```xml ############## 1. Agent初始化 ############ # agent為Agent的名字，可以隨便命名 # s1、c1、sk1是在該agent下的Source、Channel、Sink，也可以隨便命名 agent.sources = s1 agent.channels = c1 agent.sinks = sk1 ############## 2. Source配置 ############# # 設置Source的類型、端口、通道 agent.sources.s1.type = http agent.sources.s1.port = 5678 agent.sources.s1.channels = c1 ############## 3. Sink配置 ############# # 設置Sink類型為logger模式 agent.sinks.sk1.type = logger # Sink從c1通道獲取數據 agent.sinks.sk1.channel = c1 ############ 4. Channel配置 ############ # 設置Channel為內存模式 agent.channels.c1.type = memory ``` （2）啟動Agent服務 ```shell -- 切換到flume的根目錄下執行下面的語句。當然如果你已經配置了環境變量則就不需要了 bin/flume-ng agent -c conf -f myconf/http_source.conf --name agent -Dflume.root.logger=INFO,console ``` （3）發送post請求, 查看Flume輸出 ```shell curl -XPOST hadoop101:5678 -d '[{"headers":{"h1":"v1","h2":"v2"},"body":"hello body"}]' ```  <br/> # 4. taildir source 舉例 （1）創建一個`tail_source.conf`文件，編寫Agent、Source、Channel、Sink的配置信息。 ```xml ############## 1. Agent初始化 ############ # agent為Agent的名字，可以隨便命名 # s1、c1、sk1是在該agent下的Source、Channel、Sink，也可以隨便命名 agent.sources = s1 agent.channels = c1 agent.sinks = sk1 ############## 2. Source配置 ############# # 設置Source的類型 agent.sources.s1.type = TAILDIR # 定義偏移量存儲路徑 agent.sources.s1.positionFile = /opt/install/flume/myconf/taildir_position.json # 定義文件組, 多個文件 f1,f2 agent.sources.s1.filegroups = f1 f2 # 對f1指定絕對路徑 agent.sources.s1.filegroups.f1 = /opt/install/flume/myconf/f1.log # 向f1的header添加kv對 agent.sources.s1.headers.f1.headerKey1 = value1 # 對f2指定絕對路徑 agent.sources.s1.filegroups.f2 = /opt/install/flume/myconf/f2.log # 向f2的header添加kv對 agent.sources.s1.headers.f2.headerKey2 = value2 # 是否添加一個頭信息來存儲文件的絕對路徑, 默認是false agent.sources.s1.fileHeader = true # 指定通道 agent.sources.s1.channels = c1 ############## 3. Sink配置 ############# # 設置Sink類型為logger模式 agent.sinks.sk1.type = logger # Sink從c1通道獲取數據 agent.sinks.sk1.channel = c1 ############ 4. Channel配置 ############ # 設置Channel為內存模式 agent.channels.c1.type = memory ``` （2）啟動Agent服務 ```shell -- 切換到flume的根目錄下執行下面的語句。當然如果你已經配置了環境變量則就不需要了 bin/flume-ng agent -c conf -f myconf/tail_source.conf --name agent -Dflume.root.logger=INFO,console ``` 暫時不知道tail source有什么效果。