我假设每个用户自己的文件夹中的zookeeper和kafka为“kafkaflume”
kafkaflume文件夹中有两个文件夹：一个是zookeeper，另一个是kafka这里还提供了一个配置文件：flumekafka.conf
您需要根据您的要求编辑此文件。
首先需要启动zookeeper：
打开终端，转到zookeeper文件夹并启动它：

bin/zkServer.sh start

打开另一个终端，转到kafka文件夹，然后按以下方式启动： bin/kafka-server-start.sh config/server.properties 打开另一个终端，转到kafka文件夹并启动producer程序，如下所示：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic  <topicname>

现在编辑flumekafka.conf文件并准备hdfs文件夹以进行数据加载。
现在从控制台运行flume代理命令。
===========================Flume的配置文件：=======================

# Name the components on this agent

agent.sources = r1
agent.sinks =  k2
agent.channels = c1

# Describe/configure the source

agent.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.r1.zookeeperConnect = localhost:2181
agent.sources.r1.topic = <topicname>
agent.sources.r1.groupId = group1
agent.sources.r1.channels = c1
agent.sources.r1.interceptors = i1
agent.sources.r1.interceptors.i1.type = timestamp
agent.sources.r1.kafka.consumer.timeout.ms = 10000

# Describing/Configuring the sink

 agent.sinks.k2.type = hdfs
 agent.sinks.k2.hdfs.path =  hdfs://localhost:8020/user/<username>/<foldername>/%y-%m-%d
 agent.sinks.k2.hdfs.rollInterval = 5
 agent.sinks.k2.hdfs.rollSize = 0
 agent.sinks.k2.hdfs.rollCount = 0
 agent.sinks.k2.hdfs.fileType = DataStream
 agent.sinks.k2.channel = c1

# Describing/Configuring the channel agent.channels.MemChannel.type = memory

 agent.channels.c1.type=memory
 agent.channels.c1.capacity = 10000
 agent.channels.c1.transactionCapacity = 1000

谢谢你的回复，我的问题是我的en-java\u主页在文件夹名“program files”中有一些空格。