flume安装及5种采集sources

flume常用命令

//复制 文件  cp exec.conf tcp.conf

flume-ng agent -c . -f avro.conf -n a1 -Dflume.root.logger&＃61;info,console
flume-ng avro-client -c . -H linux05 -p 8888 -F /project/age.txt

flume-ng version

---------------------------------------------------------------------------------------------
linux里面不存在扩展名
1.两台机器都安装
2. 上传 解压tar -zxvf apache-flume-1.6.0-bin.tar.gz
3.vim /etc/profile
source /etc/profile
//配置环境变量

export FLUME_HOME&＃61;/root/Downloads/apache-flume-1.6.0-bin
export PATH&＃61;$PATH:$FLUME_HOME/bin

4.flume-ng version 检查是否安装成功

5.04机器 监控机器 &＃xff1a;

[root&＃64;linux05 conf]#
----------------------------
//配置文件 在/root/Downloads/apache-flume-1.6.0-bin/conf
1.Avro可以发送一个给定的文件给Flume&＃xff0c;Avro 源使用AVRO RPC机制。
创建agent配置文件  

vim avro.txt 不同的机器上
a1.sources&＃61;r1 #a1是我们要启动的Agent名字
a1.channels&＃61;c1 #sources为r1 sinks为k1 channels 为c1
a1.sinks&＃61;k1

a1.sources.r1.type&＃61;avro #指定r1的类型为AVRO
a1.sources.r1.bind&＃61;linux05 #将Source与IP地址绑定&＃xff08;这里指本机&＃xff09;
a1.sources.r1.port&＃61;8888 #指定通讯端口为4141

a1.channels.c1.type&＃61;memory #指定Channel的类型为Memory
a1.channels.c1.capacity&＃61;1000 #设置Channel的最大存储event数量为1000
a1.channels.transactionCapacity&＃61;100 #每次最大可以source中拿到或者送到sink中的event数量也是100

a1.sinks.k1.type&＃61;logger #指定k1的类型为Logger&＃xff08;不产生实体文件&＃xff0c;只在控制台显示&＃xff09;

a1.sources.r1.channels&＃61;c1
a1.sinks.k1.channel&＃61;c1
----------------------------------------------
6. //启动flume 并且 打印在控制台
flume-ng agent -c . -f avro.conf -n a1 -Dflume.root.logger&＃61;info,console

-c&＃xff1a;使用配置文件所在目录&＃xff08;这里指默认路径&＃xff0c;即$FLUME_HOME/conf&＃xff09;
-f&＃xff1a;flume定义组件的配置文件?
-n&＃xff1a;启动Agent的名称&＃xff0c;该名称在组件配置文件中定义?
-Dflume.root.logger&＃xff1a;flume自身运行状态的日志&＃xff0c;按需配置&＃xff0c;详细信息&＃xff0c;控制台打印

--------------------------------------------------------------------------------------------------------------
7.[root&＃64;linux06 conf]# flume-ng avro-client -c . -H linux05 -p 8888 -F /project/age.txt //启动客户端

&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;
2) Exec--- 监听本地文件 aa.txt
tomcat日志会打印在catalina.out 每天生成一个 tail -f catalina.out 监控 同一台机器上
1.vim exec.conf
----------------
a1.sources &＃61; r1
a1.sinks &＃61; k1
a1.channels &＃61; c1

a1.sources.r1.type &＃61; exec
a1.sources.r1.command &＃61; tail -f /aa.txt

a1.sinks.k1.type &＃61; logger

a1.channels.c1.type &＃61; memory
a1.channels.c1.capacity &＃61; 1000
a1.channels.c1.transactionCapacity &＃61; 100

a1.sources.r1.channels &＃61; c1
a1.sinks.k1.channel &＃61; c1
----------------------
2. 命令&＃xff1a;flume-ng agent -c . -f exec.conf -n a1 -Dflume.root.logger&＃61;info,console
3.[root&＃64;linux05 /]# vim aa.txt
[root&＃64;linux05 /]# echo "1111">>aa.txt

4.

3.Spool 本地模式 监听本地文件夹 同一台机器
Spool监测配置的目录下新增的文件&＃xff0c;并将文件中的数据读取出来。需要注意两点&＃xff1a;
拷贝到spool目录下的文件不可以再打开编辑。
spool目录下不可包含相应的子目录
创建agent配置文件
缺点&＃xff1a;指定文件下不能有重名(如&＃xff1a;kk.log 和 kk.log. COMPLETED),会报错 是的flume死掉

vim  spool.conf
a1.sources&＃61;r1
a1.channels&＃61;c1
a1.sinks&＃61;k1

a1.sources.r1.type&＃61;spooldir
a1.sources.r1.spoolDir&＃61;/flume

a1.channels.c1.type&＃61;memory
a1.channels.c1.capacity&＃61;1000
a1.channels.c1.transactionCapacity&＃61;100

a1.sinks.k1.type&＃61;logger

a1.sinks.k1.channel&＃61;c1
a1.sources.r1.channels&＃61;c1
---------------------------------------------------------------
[root&＃64;linux05 conf]# flume-ng agent -c . -f spool.conf -n a1 -Dflume.root.logger&＃61;info,console

[root&＃64;linux05 flume]# echo 123222 >> /flume/mm.txt

 4.Syslogtcp

Syslogtcp监听TCP的端口做为数据源

创建agent配置文件

tcp.conf   本地 监控

 //用java代码

5...JSONHandler

• 创建agent配置文件

vim http.conf //本地监控

 用java代码实现 http协议     一个用户等陆  可以显示天猫跳转淘宝  浏览器

Post方法发送json数据
----------------------------------

a1.sources &＃61; r1
a1.sinks &＃61; k1
a1.channels &＃61; c1a1.sources.r1.type &＃61; org.apache.flume.source.http.HTTPSource
a1.sources.r1.port&＃61;8888a1.sinks.k1.type &＃61; logger a1.channels.c1.type &＃61; memory a1.channels.c1.capacity &＃61; 1000 a1.channels.c1.transactionCapacity &＃61; 100 a1.sources.r1.channels &＃61; c1 a1.sinks.k1.channel &＃61; c1

----------
[root&＃64;linux05 conf]# flume-ng agent -c . -f http.conf -n a1 -Dflume.root.logger&＃61;info,console

java代码

package com.bw.tcp;import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.URL;
import java.util.Date;public class TcpTest {public static void main(String[] args) throws Exception {sendData();}public static void sendData() {try {URL url &＃61; new URL("http://linux05:8888");HttpURLConnection connection &＃61; (HttpURLConnection) url.openConnection();connection.setDoInput(true); // 设置可输入connection.setDoOutput(true); // 设置该连接是可以输出的connection.setRequestMethod("POST"); // 设置请求方式connection.setRequestProperty("Content-Type", "application/json;charset&＃61;UTF-8");PrintWriter pw &＃61; new PrintWriter(new BufferedOutputStream(connection.getOutputStream()));pw.write("[{ \"headers\" :{\"a\" : \"a1\",\"b\" : \"b1\"},\"body\" : \"test java http\"}]");pw.flush();pw.close();BufferedReader br &＃61; new BufferedReader(new InputStreamReader(connection.getInputStream(), "utf-8"));String line &＃61; null;StringBuilder result &＃61; new StringBuilder();while ((line &＃61; br.readLine()) !&＃61; null) { // 读取数据result.append(line &＃43; "\n");}connection.disconnect();System.out.println(result.toString());} catch (Exception e) {e.printStackTrace();}}}