Flume 示例

1.Syslog Tcp Source

sysylog經過配置一個端口，flume可以監控這個端口的數據。若是通往這個端口發送數據能夠被flume接收到。能夠經過socket發送。

#配置文件：syslog_case5.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
 
# Describe/configure the source
a1.sources.r1.type =syslogtcp
a1.sources.r1.port =50000
a1.sources.r1.host =192.168.233.128
a1.sources.r1.channels =c1
 
# Describe the sink
a1.sinks.k1.type = logger
 a1.sinks.k1.channel = c1
 
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

這裏咱們設置的偵聽端口爲192.168.233.128 50000

#敲命令

flume-ng agent -c -conf -f conf/syslog_case5.conf -n a1 -Dflume.root.logger=INFO,console

啓動成功後

打開另外一個終端輸入，往偵聽端口送數據

echo "hello looklook5" | nc 192.168.233.128 50000

而後看以前的終端，將會有以下顯示：

數據已經發送過來了。

也能夠經過代碼，用socket向端口發送數據

  Socket client = null;
        try {
            client = new Socket("192.168.1.75", 50000);
            OutputStream out = client.getOutputStream();
            String event = "hello world \n";
            out.write(event.getBytes());
            out.flush();
            out.close();
            client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

發送的內容要以換行符結尾"\n"

 2.Taildir Source

taildir source經過監控一個目錄或文件，目錄的文件有任何改變會將新增長的內容寫入到flume中。

a1.sources = r1
a1.sinks = k1
a1.channels = c1
 
# Describe/configure the source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /usr/apache-flume-1.7.0-bin/taildir_position1.json
a1.sources.r1.filegroups = f1
a1.sources.r1.headers.f1.headerKey1 = value1
#a1.sources.r1.filegroups.f1 = /usr/apache-flume-1.7.0-bin/test/.*
a1.sources.r1.filegroups.f1 = /opt/apache-tomcat-7.0.72/8080/logs/.*
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://127.0.0.1:9008/tomcat/%Y-%m-%d
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.filePrefix = log
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.idleTimeout = 30
 
# Use a channel which buffers events in memory
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /usr/apache-flume-1.7.0-bin/checkpoint
a1.channels.c1.dataDirs = /usr/apache-flume-1.7.0-bin/data

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1