HDFS 文件观察服务答案

【问题标题】：HDFS File Watcher ServiceHDFS 文件观察服务
【发布时间】：2020-08-01 06:12:35
【问题描述】：

我正在使用 hdfs 文件观察服务来加载配置文件，因为它在我的 flink 流作业中发生更改。

我在这里面临的问题是观察者服务正在对整个 hdfs 的变化做出反应，而不仅仅是我传递的目录。

我的代码：

public static void main( String[] args ) throws IOException, InterruptedException, MissingEventsException
  {
    HdfsAdmin admin = new HdfsAdmin( URI.create("hdfs://stage.my-org.in:8020/tmp/anurag/"), new Configuration() );
    DFSInotifyEventInputStream eventStream = admin.getInotifyEventStream();
    while( true ) {
      EventBatch events = eventStream.take();
      for( Event event : events.getEvents() ) {
        switch( event.getEventType() ) {
          case CREATE:
            System.out.print( "event type = " + event.getEventType() );
            CreateEvent createEvent = (CreateEvent) event;
            System.out.print( "  path = " + createEvent.getPath() + "\n");
            break;
          default:
            break;
        }
      }
    }
  }

程序输出：

event type = CREATE  path = /tmp/anurag/newFile.txt
event type = CREATE  path = /tmp/newFile2.txt

请帮我解决这个问题，以便我可以查看作为 URI 传递的特定目录中的文件

感谢期待

注意：如果你尝试运行这个程序，请以 hdfs 用户运行，否则你会得到 org.apache.hadoop.security.AccessControlException

【问题讨论】：

标签： java hadoop hdfs

【解决方案1】：

目前，我使用 Hadoop API 每 30 秒获取一次文件，读取它的修改时间，以及它是否大于再次重新加载文件。

【讨论】：

【解决方案2】：

InotifyEventStream 只不过是将 HDFS 事件日志解析为一个对象，无论您在构造函数中设置哪个目录，它都会将 HDFS 中的所有事件发送给您，这就是您需要运行该代码的原因之一超级组成员。

解决方案是在事件发生时对其进行过滤，仅从您想要的目录中获取这些事件。比如：

EventBatch events = eventStream.take();
ArrayList<CreateEvent> filteredEvents = new ArrayList();
for( Event event : events.getEvents() ) {
  switch( event.getEventType() ) {
    case CREATE:
      System.out.print( "event type = " + event.getEventType() );
      CreateEvent createEvent = (CreateEvent) event;
      if (createEvent.getPath() == '/your/desired/path') {
        System.out.print( "  path = " + createEvent.getPath() + "\n");
        filteredEvents.add(createEvent);
      }           
      break;
    default:
      break;
  }
}

【讨论】：