相比Flume,笔者更推荐使用Logstash做日志采集，见SODBASE CEP学习进阶篇（二）续：日志采集-Logstash、Kafka和CEP集成。如果之前项目中已经选型使用Flume，则本文供参考。
1. 启动CEP模型
启动CEP Server
./catalina.sh run
下载
loganalysis.sod
CEP模型文件。
使用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
在CEP Server的控制台输出
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产品的程序员现在可以领礼品啦