apachekafka文件读取器配置不适用于fume

lokaqttq  于 2021-06-04  发布在  Flume
关注(0)|答案(1)|浏览(527)

我是新来的flume和尝试我的第一个实验,它。我试图读取数据从一个文件使用烟雾,并将其发送到一个Kafka主题。
配置是从教程网站中提取的。配置如下所示。

a1.sources = r1
a1.sinks = sample 
a1.channels = sample-channel

a1.sources.r1.type = exec
a1.sources.r1.command = tail -f \data.txt 
a1.sources.r1.logStdErr = true
a1.channels.sample-channel.type = memory
a1.channels.sample-channel.capacity = 1000
a1.channels.sample-channel.transactionCapacity = 100
a1.sources.r1.channels = sample-channel
a1.sinks.sample.topic = sample
a1.sinks.sample.brokerList = 127.0.0.1:9092
a1.sinks.sample.requiredAcks = 1
a1.sinks.sample.batchSize = 20
a1.sinks.sample.channel = sample-channel

但这没有任何作用,它不会抛出任何错误,而是发出一些警告。日志如下所示。

2020-12-03 12:01:17,265 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateConfigFilterSet(FlumeConfiguration.java:623)] Agent configuration for 'a1' has no configfilters.
2020-12-03 12:01:17,291 (conf-file-poller-0) [WARN - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:884)] Could not configure sink  sample due to: Component has no type. Cannot configure. sample
org.apache.flume.conf.ConfigurationException: Component has no type. Cannot configure. sample
        at org.apache.flume.conf.ComponentConfiguration.configure(ComponentConfiguration.java:76)
        at org.apache.flume.conf.sink.SinkConfiguration.configure(SinkConfiguration.java:44)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:867)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:383)
        at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access$000(FlumeConfiguration.java:228)
        at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:153)
        at org.apache.flume.conf.FlumeConfiguration.<init>(FlumeConfiguration.java:133)
        at org.apache.flume.node.PropertiesFileConfigurationProvider.getFlumeConfiguration(PropertiesFileConfigurationProvider.java:194)
        at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:97)
        at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:145)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:835)

我怎样才能解决这个问题?

cvxl0en2

cvxl0en21#

正如错误所说,r1没有类型,因此它不知道如何处理源代码
你失踪了

a1.sources.r1.type = exec

就我个人而言,我建议用flebeat或telegraf over flume把文件带到Kafka

相关问题