protected def getInputFormat(conf: JobConf): InputFormat[K, V] = {
  val newInputFormat = ReflectionUtils.newInstance(inputFormatClass.asInstanceOf[Class[_]], conf)
    .asInstanceOf[InputFormat[K, V]]
  newInputFormat match {
    case c: Configurable => c.setConf(conf)
    case _ =>
  }
  newInputFormat
}