Flume 示例 - Gyoung－IT工程師數位筆記本

文章出處

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

文章列表