1.fluem 实时监测服务器端文件

  2. 1.下载插件
  3.     git clone https://github.com/keedio/flume-ftp-source.git
  5. 2.编译
  6.     mvn clean package -DskipTests
  8. 3.将生产的jar拷贝到$FLUME_HOME/lib/
  9.     cp flume-ftp-source-2.2.0.jar $FLUME_HOME/lib/
  11. 4.下载依赖jar
  12.     wget http://central.maven.org/maven2/commons-net/commons-net/3.3/commons-net-3.3.jar
  13.     wget http://central.maven.org/maven2/com/jcraft/jsch/0.1.54/jsch-0.1.54.jar
  15. 5. 将commons-net-3.3.jar 和 jsch-0.1.54.jar 拷贝到$FLUME_HOME/lib/
  16.     cp commons-net-3.3.jar $FLUME_HOME/lib/
  17.     cp jsch-0.1.54.jar $FLUME_HOME/lib/
  20. 6. 修改conf文件
  21.     touch flume-ng-ftp-source-FTP.conf
  23. 7. 添加修改的内容（flume收集文件后保存到文件中）
  24.     ## Sources Definition for agent "agent"
  25.     #ACTIVE LIST
  26.     agent.sources = ftp1
  27.     agent.sinks = k1
  28.     agent.channels = ch1 
  30.     ##### SOURCE IS ftp server
  32.     # Type of source for ftp sources
  33.     agent.sources.ftp1.type = org.keedio.flume.source.ftp.source.Source
  34.     agent.sources.ftp1.client.source = ftp
  36.     # Connection properties for ftp server
  37.     agent.sources.ftp1.name.server = 192.168.1.64
  38.     agent.sources.ftp1.port = 21
  40.     agent.sources.ftp1.user =test 
  41.     agent.sources.ftp1.password =12345 
  43.     # Process files in
  44.     agent.sources.ftp1.working.directory = /
  45.     # Proces files matches (java regex for ftp-ftps)
  46.     agent.sources.ftp1.filter.pattern = .+\\.txt
  48.     # keep file track status in folder
  49.     agent.sources.ftp1.folder = /root/download
  50.     # file track status name
  51.     agent.sources.ftp1.file.name = ftp1-status-file.ser
  53.     # Discover delay, each configured milisecond directory will be explored
  54.     agent.sources.ftp1.run.discover.delay=5000
  56.     # Process by lines
  57.     agent.sources.ftp1.flushlines = true
  59.     # Discover and process files under user's home directory
  60.     agent.sources.ftp1.search.recursive = true
  61.     # Do not process file while it is being written.
  62.     agent.sources.ftp1.processInUse = false
  63.     # If file must not be processed while it is being written, wait timeout.
  64.     agent.sources.ftp1.processInUseTimeout = 30
  67.     agent.sinks.k1.type = file_roll
  68.     agent.sinks.k1.sink.directory = /var/log/flume-ftp
  69.     agent.sinks.k1.sink.rollInterval = 7200
  71.     agent.channels.ch1.type = memory
  72.     agent.channels.ch1.capacity = 10000
  73.     agent.channels.ch1.transactionCapacity = 1000
  75.     agent.sources.ftp1.channels = ch1
  77.     agent.sinks.k1.channel = ch1
  80. 8. 运行flume-ng
  81.     ./bin/flume-ng agent -c conf -conf-file conf/flume-ng-ftp-source-FTP.conf --name agent -Dflume.root.logger=INFO,console

注意如果使用cdh的flume，需要将commons-net-3.3.jar, jsch-0.1.54.jar, flume-ftp-source-2.2.0.jar复制到一下目录

  1. cp ./* /opt/cloudera/parcels/CDH-5.14.0-1.cdh5.14.0.p0.24/lib/flume-ng/lib

如果配置flume采集后的数据发送kafka中消费

2. fpt-flume-kakfa.conf

  1. ## Sources Definition for agent "agent"
  2. #ACTIVE LIST
  3. agent.sources = ftp1
  4. agent.sinks = k1
  5. agent.channels = ch1 
  7. ##### SOURCE IS ftp server
  9. # Type of source for ftp sources
  10. agent.sources.ftp1.type = org.keedio.flume.source.ftp.source.Source
  11. agent.sources.ftp1.client.source = ftp
  13. # Connection properties for ftp server
  14. agent.sources.ftp1.name.server = 192.168.1.64
  15. agent.sources.ftp1.port = 21
  17. agent.sources.ftp1.user =test 
  18. agent.sources.ftp1.password =12345 
  20. # Process files in
  21. agent.sources.ftp1.working.directory = /
  22. # Proces files matches (java regex for ftp-ftps)
  23. agent.sources.ftp1.filter.pattern = .+\\.txt
  25. # keep file track status in folder
  26. agent.sources.ftp1.folder = /root/download
  27. # file track status name
  28. agent.sources.ftp1.file.name = ftp1-status-file.ser
  30. # Discover delay, each configured milisecond directory will be explored
  31. agent.sources.ftp1.run.discover.delay=5000
  33. # Process by lines
  34. agent.sources.ftp1.flushlines = true
  36. # Discover and process files under user's home directory
  37. agent.sources.ftp1.search.recursive = true
  38. # Do not process file while it is being written.
  39. agent.sources.ftp1.processInUse = false
  40. # If file must not be processed while it is being written, wait timeout.
  41. agent.sources.ftp1.processInUseTimeout = 30
  44. #agent.sinks.k1.type = file_roll
  45. agent.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  46. agent.sinks.kafka.topic = test
  47. agent.sinks.k1.brokerList = 192.168.1.64:9092
  49. a1.sinks.k1.kafka.flumeBatchSize = 20
  50. a1.sinks.k1.kafka.producer.acks = 1
  52. #agent.sinks.k1.sink.directory = /var/log/flume-ftp
  53. #agent.sinks.k1.sink.rollInterval = 7200
  55. agent.channels.ch1.type = memory
  56. agent.channels.ch1.capacity = 10000
  57. agent.channels.ch1.transactionCapacity = 1000
  59. agent.sources.ftp1.channels = ch1
  61. agent.sinks.k1.channel = ch1