我可以在HDFS上安装文件监视器吗?
场景: 文件不断登陆HDFS。一旦文件数量达到阈值(可以是文件数量或文件大小),我想启动一个Spark Job。
是否可以在 HDFS 上实现文件观察器来实现这一点。如果是,那么任何人都可以建议这样做的方法吗?有哪些不同的选择?Zookeeper 或 Oozie 能做到吗?
任何帮助将不胜感激。谢谢。
我可以在HDFS上安装文件监视器吗?
场景: 文件不断登陆HDFS。一旦文件数量达到阈值(可以是文件数量或文件大小),我想启动一个Spark Job。
是否可以在 HDFS 上实现文件观察器来实现这一点。如果是,那么任何人都可以建议这样做的方法吗?有哪些不同的选择?Zookeeper 或 Oozie 能做到吗?
任何帮助将不胜感激。谢谢。
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
Oozie 协调器可以做到这一点。可以根据数据可用性触发 Oozie 协调器操作。编写一个数据触发的协调器。协调器动作是基于完成标志触发的。done-flag 只是一个空文件。因此,当达到您的阈值时,将一个空文件写入目录。
旧线程...以防万一,如果有人想这样做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