SparkReader.scala

object Filters {
  def filter(filters: Seq[Column])(df: DataFrame): DataFrame = {
    filters.foldLeft(df)((df, filter) => df.filter(filter))
  }
}

trait SparkReader {
  protected def execute(reader: DataFrameReader): DataFrame

  def read(schema: Option[StructType] = None, filters: Seq[Column] = Seq.empty)(implicit sparkSession: SparkSession): DataFrame = {
    val reader = sparkSession.read
    val dataFrameReader = schema.fold(reader)(reader.schema)

    execute(dataFrameReader)
      .transform(Filters.filter(filters))
  }
}

case class SparkMongoReader(mongoUri: String) extends SparkReader {
  protected def execute(reader: DataFrameReader): DataFrame = {
    reader
      .format("com.mongodb.spark.sql.DefaultSource")
      .option("uri", mongoUri)
      .load()
  }
}