Hdfs filterHidden combinator

Author: Russell Aronson

ivory-alien-hdfs/src/main/scala/com/ambiata/ivory/alien/hdfs/Hdfs.scala

@@ -27,6 +27,9 @@ case class Hdfs[+A](action: ActionT[IO, Unit, Configuration, A]) {
 
   def |||[AA >: A](other: Hdfs[AA]): Hdfs[AA] =
     Hdfs(action ||| other.action)
+
+  def filterHidden(implicit ev: A <:< List[Path]): Hdfs[List[Path]] =
+    map(_.filter(p => !p.getName.startsWith("_") && !p.getName.startsWith(".")))
 }
 
 object Hdfs extends ActionTSupport[IO, Unit, Configuration] {

ivory-alien-hdfs/src/test/scala/com/ambiata/ivory/alien/hdfs/HdfsSpec.scala

@@ -12,6 +12,7 @@ class HdfsSpec extends Specification { def is = s2"""
    it is possible to recursively glob paths                              $e1
    can create a new dir, changing the name when it already exists        $e2
    run out of names when trying to change a dir                          $e3
+   can filter out hidden files                                           $e4
 """
 
   val basedir = "target/test/HdfsSpec/" + java.util.UUID.randomUUID()
@@ -22,7 +23,7 @@ class HdfsSpec extends Specification { def is = s2"""
     dirs.foreach(dir => new File(dir).mkdirs)
     files.foreach(f => new File(f).createNewFile)
 
-    Hdfs.globFilesRecursively(new Path(basedir)).run(new Configuration) must beOkLike(paths => paths must haveSize(4))
+    Hdfs.globFilesRecursively(new Path(basedir, "e1")).run(new Configuration) must beOkLike(paths => paths must haveSize(4))
   }
 
   def e2 = {
@@ -37,7 +38,7 @@ class HdfsSpec extends Specification { def is = s2"""
   }
 
   def e3 = {
-    val dir = new Path(basedir + "/e2/a/b/c")
+    val dir = new Path(basedir + "/e3/a/b/c")
 
     val newDir = for {
       _ <- Hdfs.mkdir(dir)
@@ -46,4 +47,16 @@ class HdfsSpec extends Specification { def is = s2"""
 
     newDir.run(new Configuration) must beOkLike(path => path must beNone)
   }
+
+  def e4 = {
+    val dir = basedir + "/e4/a/b/c"
+    val nonhidden = Seq(dir+"/f1", dir+"/f2")
+    val hidden = Seq(dir+"/.hidden", dir+"/_SUCCESS")
+    val files = nonhidden ++ hidden
+
+    new File(dir).mkdirs
+    files.foreach(f => new File(f).createNewFile)
+
+    Hdfs.globFiles(new Path(dir), "*").filterHidden.run(new Configuration) must beOkLike(paths => paths.map(_.getName) must containTheSameElementsAs(nonhidden.map(new Path(_).getName)))
+  }
 }