scala中的java重载方法

bvk5enib 于 2021-05-27 发布在 Spark

关注(0)|答案(1)|浏览(573)

我有一些密码

trait Reader {
  def read(spark: SparkSession, format: String, path: String): DataFrame
  def read[T: Encoder](spark: SparkSession, format: String, path: String): Dataset[T]
}

class LocalReader extends Reader {

  override def read[T: Encoder](spark: SparkSession, format: String, path: String): Dataset[T] = {
    spark.read
      .format(format)
      .option("header", "true")
      .load(getClass.getResource(path).getPath)
      .as[T]
  }

  override def read(spark: SparkSession, format: String, path: String): DataFrame = {
    spark.read
      .format(format)
      .option("header", "true")
      .load(getClass.getResource(path).getPath)
  }
}

object TopNSimilarCustomers extends SparkJob {
  override def appName: String = "TopNSimilarCustomers"

  override def run(spark: SparkSession, args: Array[String], reader: Reader): Unit = {

    /**
      * Only I/O here
      */

    if (args.length == 0)
      return
    val rawData = reader.read(spark, "json", "/spark-test-data.json")
    val res     = transform(spark, rawData, args(0))

  }

我有个错误 val rawData = reader.read(spark, "json", "/spark-test-data.json") 无法解析重载的方法读取。
所以我想让reader/writer用于不同的目的localreader/s3reader，因为它可以返回df和ds，所以我编写了一个重载方法，即使我必须使用一个。最终，我们必须实现这两个目标。有办法避免吗？
我怎样才能达到我想做的？有没有别的办法或者更好的办法？如何修复错误？

Java scala apache-spark overloading

来源：https://stackoverflow.com/questions/61873157/overloaded-methods-in-trait-error-spark-scala

1条答案

按热度按时间

sg24os4d1#

获取的原因 cannot resolve overloaded method read. is reader trait有两个方法，两个方法都将使用相同数量的参数。
要解决此问题，请重命名方法名称，例如 readDF & readDS 或者您也可以检查下面的代码和修改根据您的要求。

case class ReadConfig(format: String,path: String,options: Map[String,String])
    case class WriteConfig(format: String,path: String,options: Map[String,String])
    case class Config(read: ReadConfig,write: WriteConfig)

    trait Writer {
      def write(df: DataFrame): Unit
    }
    trait Reader {
      def read: DataFrame
    }

    trait RW extends Reader with Writer {
      val spark : SparkSession
      val config : Config
    }

    // Add logic for Local
    class Local(override val spark: SparkSession,override val config: Config) extends RW {
      override def read: DataFrame = {
        spark.read
          .format(config.read.format)
          .options(config.read.options)
          .load(config.read.path)
      }
      override def write(df: DataFrame): Unit = {
        df.write
          .format(config.write.format)
          .options(config.write.options)
          .save(config.write.path)
      }
    }

// Add logic for S3
 class S3(override val spark: SparkSession,override val config: Config) extends RW {
      override def read: DataFrame = {
        spark.read
          .format(config.read.format)
          .options(config.read.options)
          .load(config.read.path)
      }
      override def write(df: DataFrame): Unit = {
        df.write
          .format(config.write.format)
          .options(config.write.options)
          .save(config.write.path)
      }
    }

赞(0）回复(0）举报 2021-05-27

我来回答

scala中的java重载方法

1条答案

相关问题

热门标签

最新问答