如何在 Scala Spark 中设计一个抽象阅读器？

Question

问题如下： 我们有多种文件类型要使用自定义逻辑进行解析。假设我们有以下类型：CSV、JSON、BIN。为此，我使用 enumeratum 库创建了一个枚举

import enumeratum._
import scala.collection.immutable

/**
  * Implementation of enumeratum for various input file types.
  * See more at: https://github.com/lloydmeta/enumeratum
  */
sealed trait InputFileType extends EnumEntry

object InputFileType extends Enum[InputFileType] {

  val values: immutable.IndexedSeq[InputFileType] = findValues

  case object CSV extends InputFileType

  case object JSON extends InputFileType

  case object BIN extends InputFileType

}

对于每个InputFileType，都有一个对应的读者：

import org.apache.spark.input.PortableDataStream

class CSVReader(filename: String, file: String)
class JSONReader(filename: String, file: String)
class BINReader(filename: String, file: PortableDataStream)

由于它们应该共享一些共同的步骤并具有相似的结构，因此我想为它们创建一个超类，其中我们有一些每个读者都应该实现的步骤。但是file 输入可能会有所不同，所以我想为它制作一个 ADT。因此：

import org.apache.spark.input.PortableDataStream

sealed trait File
object File {

  case class Text(file: String) extends File

  case class BinaryStream(file: PortableDataStream) extends File

}

abstract class AbstractReader(filename: String, file: File) {
  def doStep1: Seq[String]
  def process: Result
}

class CSVReader(filename: String, file: String) extends AbstractReader(filename, Text(file)) {
  override def doStep1: Seq[String] = ???
  override def process: Result = ???
}

class JSONReader(filename: String, file: String) extends AbstractReader(filename, Text(file)) {
  override def doStep1: Seq[String] = ???
  override def process: Result = ???
}

class BINReader(filename: String, file: PortableDataStream) extends AbstractReader(filename, BinaryStream(file)) {
  override def doStep1: Seq[String] = ???
  override def process: Result = ???
}

//where
case class Data(col1: String, col2: Int)
type Result = List[Data]

现在我的想法是创建一个SparkReader 将它们读入RDD[Result]：

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

class SparkReader(spark: SparkSession, fileType: String) extends Serializable {
  def readTextFile(path: String): RDD[(String, String)] = spark.sparkContext.wholeTextFiles(path)
  def readBinaryFile(path: String): RDD[(String, PortableDataStream)] = spark.sparkContext.binaryFiles(path)
}
object SparkReader {
  def apply(spark: SparkSession, fileType: String): RDD[Result] = InputFileType.withNameInsensitive(fileType) match {
    case InputFileType.CSV => new SparkReader(spark, fileType).readTextFile(path)
      .map{case (filename: String, file: String) => new CSVReader(filename, file).process}
    case InputFileType.JSON => new SparkReader(spark, fileType).readTextFile(path)
      .map{case (filename: String, file: String) => new JSONReader(filename, file).process}
    case InputFileType.BIN => new SparkReader(spark, fileType).readBinaryFile(path)
      .map{case (filename: String, file: PortableDataStream) => new BINReader(filename, file).process}
  }
}

但是我对这个解决方案并不满意，因为最终有很多代码重复，我想要一个能够隐式决定使用哪个阅读器的设计。 最后我想实例化一个类GenericReader：

class GenericReader(spark: SparkSession, fileType: String) extends Serializable {
  def read(path: String): RDD[Result] = ???
}

然后当我调用new GenericReader(spark, "csv").read("myPath") 时，它能够得出它必须使用spark.sparkContext.wholeTextFiles 读取数据并应用CSVReader。或者当它是new GenericReader(spark, "bin").read("myPath") 时，使用spark.sparkContext.binaryFiles 创建RDD 并应用BINReader。

我也尝试应用隐式设计模式，但没有成功。 我怎样才能更优雅地解决这个问题？

Answer 1

我更喜欢使用类型类方法：

首先是示例所需的额外类型：

object types {
  type Result = List[String]
  type PortableDataStream = String
}

import types._

创建密封特征（副产品）：

sealed trait InputFileType
case object CSV extends InputFileType
case object JSON extends InputFileType
case object BIN extends InputFileType

创建您的 Reader 类型类，该类表示为带有类型参数的特征：

trait Reader[A] {
  def doStep1: Seq[String]
  def process: Result
}

创建包含“summoner”方法的伴生对象，以及在编译时可用的实例：

object Reader {
  def apply[A <: InputFileType](implicit reader: Reader[A]) = reader

  implicit object CSVReader extends Reader[CSV.type] {
    override def doStep1: Seq[String] = {
      val list = Seq("I am csvReader")
      println(list)
      list
    }

    override def process: Result = List("Result Csv")
  }

  implicit object JSONReader extends Reader[JSON.type] {
    override def doStep1: Seq[String] = {
      val list = Seq("I am jsonReader")
      println(list)
      list
    }

    override def process: Result = List("Result Json")
  }

  implicit object BINReader extends Reader[BIN.type] {
    override def doStep1: Seq[String] = {
      val list = Seq("I am binReader")
      println(list)
      list
    }

    override def process: Result = List("Result Bin")
  }

}

您的 GenericReader 可以定义为：

class GenericReader[A](implicit reader: Reader[A]) {
   def process = reader.doStep1
}

测试代码：

object SparkReader {

  import Reader._

  def main(args: Array[String]) : Unit = {
    new GenericReader[CSV.type]().process
    new GenericReader[JSON.type]().process
    new GenericReader[BIN.type]().process
  }
}

打印：

List(I am csvReader)
List(I am jsonReader)
List(I am binReader)

有很多关于如何在 Scala、Scalaz 和 Cats 中使用类型类的信息。