9

我可以在HDFS上安装文件监视器吗?

场景: 文件不断登陆HDFS。一旦文件数量达到阈值(可以是文件数量或文件大小),我想启动一个Spark Job。

是否可以在 HDFS 上实现文件观察器来实现这一点。如果是,那么任何人都可以建议这样做的方法吗?有哪些不同的选择?Zookeeper 或 Oozie 能做到吗?

任何帮助将不胜感激。谢谢。

4

3 回答 3

14

Hadoop 2.6 引入DFSInotifyEventInputStream了您可以用于此目的。您可以从中获取它的一个实例,HdfsAdmin然后只需调用.take().poll()获取所有事件。事件类型包括删除、追加和创建,它们应该涵盖您要查找的内容。

这是一个基本的例子。确保以hdfs用户身份运行它,因为管理界面需要 HDFS root。

public static void main( String[] args ) throws IOException, InterruptedException, MissingEventsException
{
    HdfsAdmin admin = new HdfsAdmin( URI.create( args[0] ), new Configuration() );
    DFSInotifyEventInputStream eventStream = admin.getInotifyEventStream();
    while( true ) {
        EventBatch events = eventStream.take();
        for( Event event : events.getEvents() ) {
            System.out.println( "event type = " + event.getEventType() );
            switch( event.getEventType() ) {
                case CREATE:
                    CreateEvent createEvent = (CreateEvent) event;
                    System.out.println( "  path = " + createEvent.getPath() );
                    break;
                default:
                    break;
            }
        }
    }
}

这是一篇更详细地介绍它的博客文章:

http://johnjianfang.blogspot.com/2015/03/hdfs-6634-inotify-in-hdfs.html?m=1

于 2015-05-01T17:55:56.913 回答
1

Oozie 协调器可以做到这一点。可以根据数据可用性触发 Oozie 协调器操作。编写一个数据触发的协调器。协调器动作是基于完成标志触发的。done-flag 只是一个空文件。因此,当达到您的阈值时,将一个空文件写入目录。

于 2017-06-05T13:22:57.013 回答
0

旧线程...以防万一,如果有人想这样做Scala

import java.net.URI

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hdfs.client.HdfsAdmin
import org.apache.hadoop.hdfs.inotify.Event.{AppendEvent, CreateEvent, RenameEvent}


object HDFSTest extends App {
  val admin = new HdfsAdmin( URI.create( "hdfs://namenode:port" ), new Configuration() )
  val eventStream = admin.getInotifyEventStream()

  while( true ) {
    val events =  eventStream.poll(2l, java.util.concurrent.TimeUnit.SECONDS)
    events.getEvents.toList.foreach { event ⇒
      println(s"event type = ${event.getEventType}")
      event match {
        case create: CreateEvent ⇒
          println("CREATE: " + create.getPath)

        case rename: RenameEvent ⇒
          println("RENAME: " + rename.getSrcPath + " => " + rename.getDstPath)

        case append: AppendEvent ⇒
          println("APPEND: " + append.getPath)

        case other ⇒
          println("other: " + other)
      }
    }
  }
}

万一,如果想使用模拟用户...设置环境变量:HADOOP_USER_NAME=user-name

于 2019-09-10T20:45:19.243 回答