:-: **常用的Sink類型** | 類型 | 描述 | | --- | --- | | avro sink | 作為avro客戶端向avro服務端發送avro事件。 | | HDFS sink | 將事件寫入Hadoop分布式文件系統（HDFS）。 | | Hive sink | 包含分隔文本或JSON數據流事件直接進入Hive表或分區；<br/>傳入的事件數據字段映射到Hive表中相應的列。 | | HBase sink | 此 Sink 將數據寫入 HBase。 | | Kafka sink | 此 Sink 將數據寫入 Kafka。 | <br/> **怎么使用？** （1）查看官方文檔：https://flume.apache.org/releases/content/1.8.0/FlumeUserGuide.html  <br/> [TOC] # 1. http和avro的配合進行PRC（遠程）調用 （1）編寫Flume文件，設置Agent、Source、Channel、Sink的配置信息。 *`http_source.conf`* ```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. Channel配置 ############ # 設置Channel為內存模式 agent.channels.c1.type = memory ############ 4. Sink配置 ############# # 指定sink類型、主機、端口號（不能與Source的端口號相同）、通道 agent.sinks.sk1.type = avro agent.sinks.sk1.hostname = hadoop101 agent.sinks.sk1.port = 4444 agent.sinks.sk1.channel = c1 ``` <br/> *`avro_source.conf`* ```xml ############## 1. Agent初始化 ############ # agent為Agent的名字，可以隨便命名 # s1、c1、sk1是在該agent下的Source、Channel、Sink，也可以隨便命名 agent.sources = s1 agent.channels = c1 agent.sinks = sk1 ############## 2. Source配置 ############# # 指定Source類型、主機、端口號（與上面的sink端口號相同）、通道 agent.sources.s1.type = avro agent.sources.s1.bind = hadoop101 agent.sources.s1.port = 4444 agent.sources.s1.channels = c1 ############ 3. Channel配置 ############ # 指定Channel類型 agent.channels.c1.type = memory ############ 4. Sink配置 ############# # 指定sink類型、通道 agent.sinks.sk1.type = logger agent.sinks.sk1.channel = c1 ``` （2）啟動Flume ```shell -- 切換到flume的根目執行語句。當然如果你已經配置了環境變量則就不需要了 -- 先啟動avro_source bin/flume-ng agent -c conf -f myconf/avro_source.conf --name agent -Dflume.root.logger=INFO,console -- 再啟動http_source bin/flume-ng agent -c conf -f myconf/http_source.conf --name agent -Dflume.root.logger=INFO,console ```   （3）發送post請求，查看avro_source的Flume輸出 ```shell curl -XPOST hadoop101:5678 -d '[{"headers":{"h1":"v1","h2":"v2"},"body":"hello body"}]' ```  <br/> # 2. HDFS Sink的使用 （1）編寫`hdfs_sink.conf`文件 ```conf ############## 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 = netcat agent.sources.s1.bind = hadoop101 agent.sources.s1.port = 5678 agent.sources.s1.channels = c1 ############ 3. Channel配置 ############ # 設置Channel為內存模式 agent.channels.c1.type = memory ############ 4. Sink配置 ############# # 設置Sink類型 agent.sinks.sk1.type = hdfs agent.sinks.sk1.hdfs.path = /flume/data/ # Sink從c1通道獲取數據 agent.sinks.sk1.channel = c1 ``` （2）啟動Flume ```shell [root@hadoop101 flume]# bin/flume-ng agent -c conf -f myconf/hdfs_sink.conf --name agent -Dflume.root.logger=INFO,console ``` （3）啟動telnet發送數據 ```shell [root@hadoop101 flume]# telnet hadoop101 5678 Trying 192.168.75.129... Connected to hadoop101. Escape character is '^]'. hello world OK hello spark OK hello hadoop scala OK ``` （4）查看Flume日志輸出  （5）查看hdfs上的文件 ```shell [root@hadoop101 hadoop]# bin/hdfs dfs -text /flume/data/FlumeData.1611000631177 21/01/19 04:15:33 WARN util.NativeCodeLoader: Unable to load 1611000632253 68 65 6c 6c 6f 20 77 6f 72 6c 64 0d 1611000636996 68 65 6c 6c 6f 20 73 70 61 72 6b 0d 1611000644003 68 65 6c 6c 6f 20 68 61 64 6f 6f 70 20 73 63 61 6c 61 0d ```