flume-ng-extends-source
flume-ng-extends-source copied to clipboard
ningg
No channels set for spoolDirTailFile
用1.6版本的flume，使用这个jar包，报org.apache.flume.conf.ConfigurationException: No channels set for spoolDirTailFile，配置channels之后（agent.sources.spoolDirTailTile.channels = c1），还是不行，请指导。
agent.sources = spoolDirTailFile agent.channels = dir2kafkachannel agent.sinks = kafkasink
#pooling dir and tail file Source agent.sources.spoolDirTailFile.type = com.github.ningg.flume.source.SpoolDirectoryTailFileSource #on WIN plantform spoolDir should be format like: E:/program files/spoolDir #Note: the value of spoolDir MUST NOT be surrounded by quotation marks. agent.sources.spoolDirTailFile.spoolDir = /var/data-flume-bt/test agent.sources.spoolDirTailFile.fileSuffix = .COMPLETED agent.sources.spoolDirTailFile.deletePolicy = never agent.sources.spoolDirTailFile.ignorePattern = ^$ agent.sources.spoolDirTailFile.targetPattern = .(\d){4}(\d){2}(\d){2}(\d){2}. agent.sources.spoolDirTailFile.targetFilename = yyyyMMddHH agent.sources.spoolDirTailFile.trackerDir = .flumespooltail agent.sources.spoolDirTailFile.consumeOrder = oldest agent.sources.spoolDirTailFile.batchSize = 100 agent.sources.spoolDirTailFile.inputCharset = UTF-8 agent.sources.spoolDirTailFile.decodeErrorPolicy = REPLACE agent.sources.spoolDirTailFile.deserializer = LINE agent.sources.spoolDirTailTile.channels=dir2kafkachannel agent.sources.spoolDirTailTile.fileHeader = true
agent.channels.dir2kafkachannel.type = memory agent.channels.dir2kafkachannel.capacity = 10000 agent.channels.dir2kafkachannel.transactionCapacity = 10000
agent.sinks.kafkasink.type = org.apache.flume.sink.kafka.KafkaSink agent.sinks.kafkasink.topic = test-topic17 agent.sinks.kafkasink.partition.key = 0 agent.sinks.kafkasink.brokerList = localhost:9092 agent.sinks.kafkasink.requiredAcks = 0 agent.sinks.kafkasink.batchSize = 500 agent.sinks.kafkasink.channel = dir2kafkachannel
不好意思，刚贴的格式有点乱。这是完整的配置，谢谢
[root@lgb apache-flume-1.6.0-bin]# bin/flume-ng agent --conf conf --conf-file conf/flume-conf-bt.conf --name agent -Dflume.root.logger=INFO,console Info: Including Hive libraries found via () for Hive access
- exec /usr/java/jdk1.7.0_60/bin/java -Xms4096m -Xmx4096m -Dflume.root.logger=INFO,console -cp '/usr/local/apache-flume-1.6.0-bin/conf:/usr/local/apache-flume-1.6.0-bin/lib/:/lib/' -Djava.library.path= org.apache.flume.node.Application --conf-file conf/flume-conf-bt.conf --name agent
2016-02-04 16:28:20,162 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:61)] Configuration provider starting
2016-02-04 16:28:20,175 (conf-file-poller-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:133)] Reloading configuration file:conf/flume-conf-bt.conf
2016-02-04 16:28:20,189 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,191 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,192 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,192 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:931)] Added sinks: kafkasink Agent: agent
2016-02-04 16:28:20,193 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,193 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,193 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,194 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)] Processing:kafkasink
2016-02-04 16:28:20,212 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSources(FlumeConfiguration.java:590)] Could not configure source spoolDirTailFile due to: Failed to configure component!
org.apache.flume.conf.ConfigurationException: Failed to configure component!
at org.apache.flume.conf.source.SourceConfiguration.configure(SourceConfiguration.java:111)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSources(FlumeConfiguration.java:567)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:346)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access$000(FlumeConfiguration.java:213)
at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:127)
at org.apache.flume.conf.FlumeConfiguration.
(FlumeConfiguration.java:109) at org.apache.flume.node.PropertiesFileConfigurationProvider.getFlumeConfiguration(PropertiesFileConfigurationProvider.java:189) at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:89) at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.flume.conf.ConfigurationException: No channels set for spoolDirTailFile at org.apache.flume.conf.source.SourceConfiguration.configure(SourceConfiguration.java:69) ... 15 more 2016-02-04 16:28:20,220 (conf-file-poller-0) [INFO - org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:141)] Post-validation flume configuration contains configuration for agents: [agent] 2016-02-04 16:28:20,221 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:145)] Creating channels 2016-02-04 16:28:20,241 (conf-file-poller-0) [INFO - org.apache.flume.channel.DefaultChannelFactory.create(DefaultChannelFactory.java:42)] Creating instance of channel dir2kafkachannel type memory 2016-02-04 16:28:20,251 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:200)] Created channel dir2kafkachannel 2016-02-04 16:28:20,253 (conf-file-poller-0) [INFO - org.apache.flume.sink.DefaultSinkFactory.create(DefaultSinkFactory.java:42)] Creating instance of sink: kafkasink, type: org.apache.flume.sink.kafka.KafkaSink 2016-02-04 16:28:20,264 (conf-file-poller-0) [INFO - org.apache.flume.sink.kafka.KafkaSink.configure(KafkaSink.java:209)] Using the static topic: test-topic17 this may be over-ridden by event headers 2016-02-04 16:28:20,266 (conf-file-poller-0) [INFO - org.apache.flume.sink.kafka.KafkaSinkUtil.getKafkaProperties(KafkaSinkUtil.java:34)] context={ parameters:{topic=test-topic17, partition.key=0, brokerList=localhost:9092, requiredAcks=0, batchSize=500, type=org.apache.flume.sink.kafka.KafkaSink, channel=dir2kafkachannel} } 2016-02-04 16:28:20,308 (conf-file-poller-0) [INFO - org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:114)] Channel dir2kafkachannel connected to [kafkasink] 2016-02-04 16:28:20,324 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:138)] Starting new configuration:{ sourceRunners:{} sinkRunners:{kafkasink=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@49a1afb1 counterGroup:{ name:null counters:{} } }} channels:{dir2kafkachannel=org.apache.flume.channel.MemoryChannel{name: dir2kafkachannel}} } 2016-02-04 16:28:20,326 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:145)] Starting Channel dir2kafkachannel 2016-02-04 16:28:20,433 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)] Monitored counter group for type: CHANNEL, name: dir2kafkachannel: Successfully registered new MBean. 2016-02-04 16:28:20,433 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)] Component type: CHANNEL, name: dir2kafkachannel started 2016-02-04 16:28:20,434 (conf-file-poller-0) [INFO - org.apache.flume.node.Application.startAllComponents(Application.java:173)] Starting Sink kafkasink 2016-02-04 16:28:21,008 (lifecycleSupervisor-1-0) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Verifying properties 2016-02-04 16:28:21,065 (lifecycleSupervisor-1-0) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Property key.serializer.class is overridden to kafka.serializer.StringEncoder 2016-02-04 16:28:21,065 (lifecycleSupervisor-1-0) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Property metadata.broker.list is overridden to localhost:9092 2016-02-04 16:28:21,066 (lifecycleSupervisor-1-0) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Property request.required.acks is overridden to 0 2016-02-04 16:28:21,066 (lifecycleSupervisor-1-0) [INFO - kafka.utils.Logging$class.info(Logging.scala:68)] Property serializer.class is overridden to kafka.serializer.DefaultEncoder 2016-02-04 16:28:21,136 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)] Monitored counter group for type: SINK, name: kafkasink: Successfully registered new MBean. 2016-02-04 16:28:21,136 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)] Component type: SINK, name: kafkasink started
这个是整个启动之后的日志，请阅知，谢谢。
2016-02-04 16:28:20,212 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSources(FlumeConfiguration.java:590)] Could not configure source spoolDirTailFile due to: Failed to configure component!
org.apache.flume.conf.ConfigurationException: Failed to configure component!
at org.apache.flume.conf.source.SourceConfiguration.configure(SourceConfiguration.java:111)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSources(FlumeConfiguration.java:567)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:346)
at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access$000(FlumeConfiguration.java:213)
at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:127)
at org.apache.flume.conf.FlumeConfiguration.
这个是错误日志。
修改一下配置文件：
agent.sources.spoolDirTailTile.channels=dir2kafkachannel
修改为：
agent.sources.spoolDirTailTile.channels = dir2kafkachannel
调整一下 source、sink、channel 之间对应关系的位置，把下面的配置，放置到配置文件末尾：
agent.sources.spoolDirTailTile.channels = dir2kafkachannel agent.sinks.kafkasink.channel = dir2kafkachannel
估计你 spoolDirTailFile 的配置有错，我刚刚细看了一边启动日志：channel、sink都成功启动了，就是 spoolDirTailFile 这个source 启动时，说配置不对。