如何仅监控指定HDFS目录及其子目录的文件创建事件？

我有一个会定期生成文件的HDFS目录，需要在其子目录有文件创建时收到通知，以便从该HDFS位置读取文件内容并发送至Kafka Topic。我了解到可通过HDFS iNotify实现此功能，但它存在弊端：会监控整个HDFS的所有事件，而非仅针对指定目录。

以下是我当前的示例代码：

public class HdfsINotifyExample {
    public static void main(String[] args) throws IOException, InterruptedException, MissingEventsException {
        long lastReadTxid = 0;
        if (args.length > 1) {
            lastReadTxid = Long.parseLong(args[1]);
        }
        System.out.println("lastReadTxid = " + lastReadTxid);
        HdfsAdmin admin = new HdfsAdmin(URI.create(args[0]), new Configuration());
        DFSInotifyEventInputStream eventStream = admin.getInotifyEventStream(lastReadTxid);
        while (true) {
            EventBatch batch = eventStream.take();
            System.out.println("TxId = " + batch.getTxid());
            for (Event event : batch.getEvents()) {
                System.out.println("event type = " + event.getEventType());
                switch (event.getEventType()) {
                    case CREATE:
                        CreateEvent createEvent = (CreateEvent) event;
                        System.out.println(" path = " + createEvent.getPath());
                        System.out.println(" owner = " + createEvent.getOwnerName());
                        System.out.println(" ctime = " + createEvent.getCtime());
                        break;
                    default:
                        break;
                }
            }
        }
    }
}

请问是否有更优方式，仅监控指定HDFS目录下的文件创建事件，而非所有事件类型？

当然有办法！虽然HDFS iNotify本身没有提供内置的目录过滤机制（它会推送集群内所有的HDFS事件），但我们可以在应用层添加精准的过滤逻辑，只处理你关心的指定目录下的文件创建事件。这是最直接且高效的优化方式，具体实现如下：

1. 添加事件与路径过滤逻辑 #

我们可以在代码中定义要监控的目标目录，然后在事件循环中：

• 只处理CREATE类型的事件（直接跳过其他事件类型）
• 验证事件对应的文件路径是否属于目标目录（或其子目录）

修改后的代码示例：

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.client.HdfsAdmin;
import org.apache.hadoop.hdfs.inotify.Event;
import org.apache.hadoop.hdfs.inotify.EventBatch;
import org.apache.hadoop.hdfs.inotify.MissingEventsException;
import org.apache.hadoop.hdfs.inotify.CreateEvent;
import org.apache.hadoop.conf.Configuration;
import java.io.IOException;
import java.net.URI;

public class HdfsTargetDirINotify {
    // 定义要监控的目标HDFS目录（替换为你的实际路径）
    private static final String MONITORED_DIR = "/user/your-target-dir";

    public static void main(String[] args) throws IOException, InterruptedException, MissingEventsException {
        long lastReadTxid = 0;
        if (args.length > 1) {
            lastReadTxid = Long.parseLong(args[1]);
        }
        System.out.println("lastReadTxid = " + lastReadTxid);

        HdfsAdmin admin = new HdfsAdmin(URI.create(args[0]), new Configuration());
        DFSInotifyEventInputStream eventStream = admin.getInotifyEventStream(lastReadTxid);
        
        Path monitoredPath = new Path(MONITORED_DIR);

        while (true) {
            EventBatch batch = eventStream.take();
            System.out.println("TxId = " + batch.getTxid());

            for (Event event : batch.getEvents()) {
                // 只处理CREATE类型的事件，直接跳过其他类型
                if (event.getEventType() != Event.EventType.CREATE) {
                    continue;
                }

                CreateEvent createEvent = (CreateEvent) event;
                Path eventPath = new Path(createEvent.getPath());
                
                // 验证路径是否属于目标目录（包括子目录）
                if (eventPath.startsWith(monitoredPath)) {
                    // 这里添加你的业务逻辑：读取文件内容并发送到Kafka
                    System.out.println("捕获到目标目录下的文件创建事件：");
                    System.out.println(" 路径 = " + createEvent.getPath());
                    System.out.println(" 所有者 = " + createEvent.getOwnerName());
                    System.out.println(" 创建时间 = " + createEvent.getCtime());
                    
                    // TODO: 读取文件内容并发送至Kafka Topic
                }
            }
        }
    }
}

2. 关键优化点说明 #

• 路径匹配的可靠性：使用HDFS的Path类进行路径比较，而不是直接用字符串匹配，这样可以自动处理路径中的斜杠、相对路径等细节问题，避免匹配错误。
• 事件类型过滤：直接跳过非CREATE类型的事件，减少不必要的逻辑处理，提升性能。
• 灵活的目录监控：如果需要监控多个目录，可以将MONITORED_DIR改为一个Set<Path>集合，然后循环判断事件路径是否属于其中任意一个目录。
• TxId持久化：记得将lastReadTxid持久化到外部存储（比如数据库、文件），否则应用重启后会重新处理之前已经消费过的事件，导致重复发送Kafka消息。

3. 其他可选方案 #

如果你的场景需要更复杂的监控逻辑，也可以考虑：

• Apache Flume的HDFS Source：Flume内置了HDFS目录监控功能，可以配置只监控指定目录，并将文件内容发送到Kafka Sink，无需自己编写iNotify代码。
• HDFS Snapshot结合定时扫描：如果对实时性要求不高，可以定期扫描目标目录的快照，对比差异来发现新创建的文件，但这种方式实时性不如iNotify。

内容的提问来源于stack exchange，提问作者Hema12