如何实现在Hadoop分布式上用Spooling Directory Source监听目录收集日志，之后用hdfs sink上传进hdfs？

要使用Flume进行数据日志采集，在Hadoop分布式环境中将日志文件上传到HDFS，可以按照以下步骤进行操作：

首先，确保你已经正确安装和配置了Flume和Hadoop，并能够正常运行。确保Flume的环境变量已设置好，并且Hadoop集群已经启动。

在Flume的配置文件中，定义一个Spooling Directory Source来监听目录并收集日志。同时，配置一个HDFS Sink来将日志文件上传到HDFS。

示例配置文件（flume.conf）如下：

使用命名的组件，可以为Source、Sink和Channel命名

agent.sources = spooling-source
agent.sinks = hdfs-sink
agent.channels = memory-channel

定义 Spooling Directory Source

agent.sources.spooling-source.type = spooldir
agent.sources.spooling-source.spoolDir = /path/to/log/directory
agent.sources.spooling-source.fileHeader = true

定义 HDFS Sink

agent.sinks.hdfs-sink.type = hdfs
agent.sinks.hdfs-sink.hdfs.path = hdfs://namenode:8020/path/to/hdfs/directory
agent.sinks.hdfs-sink.hdfs.fileType = DataStream
agent.sinks.hdfs-sink.hdfs.rollSize = 0
agent.sinks.hdfs-sink.hdfs.rollCount = 10000
agent.sinks.hdfs-sink.hdfs.rollInterval = 600

定义 Memory Channel

agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000
agent.channels.memory-channel.transactionCapacity = 1000

组件之间的连接关系

agent.sources.spooling-source.channels = memory-channel
agent.sinks.hdfs-sink.channel = memory-channel

在上述配置中，将/path/to/log/directory替换为要监听的目录路径，并将hdfs://namenode:8020/path/to/hdfs/directory替换为要上传到的HDFS目录路径。你还可以根据需要进行其他配置，如文件滚动大小、文件滚动计数和文件滚动时间间隔等。

启动Flume代理并运行数据采集任务：

$ flume-ng agent -n agent -f /path/to/flume.conf -c /path/to/flume_agents_dir -Dflume.root.logger=INFO,console

其中，/path/to/flume.conf是上述配置文件的路径，/path/to/flume_agents_dir是Flume代理的目录路径。

Flume将开始监听指定目录的日志文件，并将其上传到HDFS中的指定目录。

这样，你就可以使用Flume的Spooling Directory Source来监听目录，并使用HDFS Sink将日志文件上传到HDFS中了。请根据你的实际环境和需求修改配置文件中的路径和参数。