使用 Flume 将数据从 kafka 摄取到 HDFS:: ConfigurationException: Bootstrap Servers must be specified

Question

我正在尝试使用从 kafka 源到 hdfs 的水槽摄取数据。下面是我的水槽 conf 文件。

flume1.sources = kafka-source-1
flume1.channels = hdfs-channel-1
flume1.sinks = hdfs-sink-1
flume1.sources.kafka-source-1.type = org.apache.flume.source.kafka.KafkaSource
flume1.sources.kafka-source-1.bootstrap.servers = localhost:9092
flume1.sources.kafka-source-1.zookeeperConnect = localhost:2181
flume1.sources.kafka-source-1.topic = MYNEWSFEEDS
flume1.sources.kafka-source-1.batchSize = 100
flume1.sources.kafka-source-1.channels = hdfs-channel-1
flume1.channels.hdfs-channel-1.type   = memory
flume1.sinks.hdfs-sink-1.channel = hdfs-channel-1
flume1.sinks.hdfs-sink-1.type = hdfs
flume1.sinks.hdfs-sink-1.hdfs.writeFormat = Text
flume1.sinks.hdfs-sink-1.hdfs.fileType = DataStream
flume1.sinks.hdfs-sink-1.hdfs.filePrefix = test-events
flume1.sinks.hdfs-sink-1.hdfs.useLocalTimeStamp = true
flume1.sinks.hdfs-sink-1.hdfs.path = hdfs://quickstart.cloudera:8020/tmp
flume1.sinks.hdfs-sink-1.hdfs.rollCount=100
flume1.sinks.hdfs-sink-1.hdfs.rollSize=0
flume1.channels.hdfs-channel-1.capacity = 10000
flume1.channels.hdfs-channel-1.transactionCapacity = 1000

我正在使用以下命令来运行水槽代理：

sudo flume-ng agent --name flume1 --conf-file '/etc/flume-ng/conf/flafka.conf' Dflume.root.logger=TRACE,console

但我遇到以下错误：

18/03/12 16:49:18 错误节点。AbstractConfigurationProvider：来源 kafka-source-1 由于配置过程中的错误已被删除 org.apache.flume.conf.ConfigurationException：引导服务器必须 指定在 org.apache.flume.source.kafka.KafkaSource.doConfigure(KafkaSource.java:330) 在 org.apache.flume.source.BasicSourceSemantics.configure(BasicSourceSemantics.java:65) 在 org.apache.flume.source.AbstractPollableSource.configure(AbstractPollableSource.java:63) 在 org.apache.flume.conf.Configurables.configure(Configurables.java:41) 在 org.apache.flume.node.AbstractConfigurationProvider.loadSources(AbstractConfigurationProvider.java:326) 在 org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:97) 在 org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140) 在 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) 在 java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 在 java.lang.Thread.run(Thread.java:745)

虽然，我已经在 conf 文件中指定了引导服务器，但它仍然给出了同样的错误。尝试了很多排列组合，但都没有成功。

Answer 1

根据official JavaDoc，你应该替换

flume1.sources.kafka-source-1.bootstrap.servers = localhost:9092

与

flume1.sources.kafka-source-1.kafka.bootstrap.servers = localhost:9092