相比Flume,筆者更推薦使用Logstash做日誌採集，見SODBASE CEP學習進階篇（二）續：日誌採集-Logstash、Kafka和CEP整合。如果之前專案中已經選型使用Flume，則本文供參考。

1. 啟動CEP模型

啟動CEP Server

./catalina.sh run

使用Server Admin將loganalysis.soddata2安裝到CEP Server，並啟動，控制檯輸出

May 29, 2016 5:14:26 PM com.sodbase.cep.graphmodelexecutor.GraphModelExecutorImpl resetGraphModelInstatnce
INFO: com.sodbase.outputadaptor.PrintEventOutputAdaptor
'start cep model CEPModelPrimaryKey [modelname=loganalysis, modelspacename=admin, version=1.0, serveraddress=localhost:16111] startnothotswap','admin','2016-05-29 17:14:26
 

2. 配置flume

下載apache-flume-1.5.2-bin.zip，解壓自定義目錄。將SODBASE Studio lib目錄下的sodbase-cep-engine.jar,sodbase-studio.jar,sodbase-dataadaptor-socket.jar,sodbase-dataadaptor-flume.jar拷貝到flume的lib目錄下。

在flume的conf目錄下編輯配置檔案

$ vi syslog_tcp.conf

a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = com.sodbase.dataadaptor.flume.CEPServerSink
a1.sinks.k1.channel = memoryChannel
#修改為CEP Server的地址，埠與loganalysis的輸入中配置的埠一致
a1.sinks.k1.CEPServerSocketIpPort=localhost:12345
a1.sinks.k1.CEPServerSocketRetryNum=2
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
 

執行flume採集,在flume的bin目錄下

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

3. 測試

$ echo "hello idoall.org syslog" | nc localhost 5140

Sun May 29 17:18:41 CST 2016 T1 flumeeventdata: d1,d2,d3 Sun May 29 17:18:41 CST 2016 

Sun May 29 17:18:58 CST 2016 T1 flumeeventdata: d4,d5,d6 Sun May 29 17:18:58 CST 2016 
 

4. UDP syslog

在flume配置檔案中，修改

a1.sources.r1.type = syslogudp
a1.sources.r1.host = 0.0.0.0

測試時使用

$ echo "hello idoall.org syslog" | nc -u localhost 5140

開發者社群活動，使用SODBASE產品的程式設計師現在可以領禮品啦