文件 reader 配置不适用于 Fume

Question

我是 flume 的新手，正在尝试使用 it.Iam 尝试使用 fume 从文件中读取数据并将其发送到 kafka 主题。

配置来自教程website。配置如下所示。

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

但这不起作用 anything.It 没有抛出任何错误，而是一些警告。日志如下所示。

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[=12=]0(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)

我该如何解决这个问题？

Answer 1

如错误所述，r1 没有类型，因此它不知道如何处理您的源

你不见了

a1.sources.r1.type = exec

就我个人而言，我建议 Flume 使用 Filebeat 或 Telegraf 将文件传输到 Kafka

文件 reader 配置不适用于 Fume

File reader configuration doesn't work for Fume

apache-kafka

flume

flume-ng