repartition small files on hdfs

repackage.scala

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.SparkSession


object App {

  def lsFiles(path: String, ss: SparkSession) = {
    FileSystem.get(ss.sparkContext.hadoopConfiguration).listStatus(new Path(path)).toSeq.map {
      _.getPath.toString
    }
  }

  def repackageFile(path: String, out: String, ss: SparkSession) {
    ss.read.textFile(path).coalesce(50).write.option("compression", "snappy").text(out)
  }

  def process(path: String, outPrefix: String, ss: SparkSession) = {
    lsFiles(path, ss).foreach { sPath =>
      val subDir = lsFiles(sPath, ss)
      if (subDir.size > 10) {
        println(s"repackaging $subDir")
        val output = outPrefix + sPath.split("/").last
        repackageFile(sPath, output, ss)
      }
    }
  }

}