scala - 到处导入 Spark 隐式的解决方法

Question

我是 Spark 2.0 的新手，并且在我们的代码库中使用数据集。我有点注意到我需要在代码中的任何地方导入spark.implicits._。例如:

File A
class A {
    def job(spark: SparkSession) = {
        import spark.implcits._
        //create dataset ds
        val b = new B(spark)
        b.doSomething(ds)
        doSomething(ds)
    }
    private def doSomething(ds: Dataset[Foo], spark: SparkSession) = {
        import spark.implicits._
        ds.map(e => 1)            
    }
}

File B
class B(spark: SparkSession) {
    def doSomething(ds: Dataset[Foo]) = {
        import spark.implicits._
        ds.map(e => "SomeString")
    }
}

我想问的是是否有更干净的方法可以做到

ds.map(e => "SomeString")

无需在我执行映射的每个函数中导​​入隐式内容？如果我不导入它，则会收到以下错误:

Error:(53, 13) Unable to find encoder for type stored in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases.

Answer 1

有一点帮助的是在类或对象而不是每个函数中进行导入。对于“文件 A”和“文件 B”示例:

File A
class A {
    val spark = SparkSession.builder.getOrCreate()
    import spark.implicits._

    def job() = {
        //create dataset ds
        val b = new B(spark)
        b.doSomething(ds)
        doSomething(ds)
    }

    private def doSomething(ds: Dataset[Foo]) = {
        ds.map(e => 1)            
    }
}

File B
class B(spark: SparkSession) {
    import spark.implicits._

    def doSomething(ds: Dataset[Foo]) = {    
        ds.map(e => "SomeString")
    }
}

通过这种方式，您可以获得可管理数量的导入。

不幸的是，据我所知，没有其他方法可以进一步减少进口数量。这是因为在执行实际的导入时需要SparkSession对象。因此，这是可以做的最好的事情。

<小时/>

更新:

更方便的方法是创建 Scala Trait 并将其与空的 Object 组合。这允许在每个文件的顶部轻松导入隐式，同时允许扩展特征以使用 SparkSession 对象。

示例:

trait SparkJob {
  val spark: SparkSession = SparkSession.builder.
    .master(...)
    .config(..., ....) // Any settings to be applied
    .getOrCreate()
}

object SparkJob extends SparkJob {}

这样我们就可以对文件 A 和 B 执行以下操作:

文件A:

import SparkJob.spark.implicits._
class A extends SparkJob {
  spark.sql(...) // Allows for usage of the SparkSession inside the class
  ...
}

文件B:

import SparkJob.spark.implicits._
class B extends SparkJob {
  ...    
}

请注意，只需为使用 spark 对象本身的类或对象扩展 SparkJob 即可。