如何做Spark 版本兼容-阿里云开发者社区

如何做Spark 版本兼容

2016-09-12 3697

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 我们知道Spark2.0 ，Spark 1.6还有Spark 1.5 三者之间版本是不兼容的，尤其是一些内部API变化比较大。如果你的系统使用了不少底层的API,那么这篇文章或许对你有帮助。我们介绍的兼容相关一些技巧，主要包括动态编译以及反射等方式，也用到了Scala的一些语言特性。

案例

在Spark 1.6 时，大部分机器学习相关的类使用的向量还是

org.apache.spark.mllib.linalg.Vector
        
          
        
        
        
          
          AI 代码解读

而到2.0后，已经基本都变更成

org.apache.spark.ml.linalg.Vector
        
          
        
        
        
          
          AI 代码解读

同理对应的Vectors object 也是。这就造成了一个比较大的困难，比如下面的代码就很难做到兼容了，切换Spark就无法通过编译：

//定义一个函数，将一个字符串转化为Vector
val t = udf { (features: String) =>

      if (!features.contains(":")) {
        val v = features.split(",|\\s+").map(_.toDouble)
        Vectors.dense(v)
      } else {
        val v = features.split(",|\\s+").map(_.split(":")).map(f => (f(0).toInt, f(1).toDouble))
        Vectors.sparse(vectorSize, v)
      }

    }

//Dataframe转化为只有label 和 features列。
training.select(
      col("label") cast (org.apache.spark.sql.types.DoubleType),
      t(col("features")) as "features"
)
        
          
        
        
        
          
          AI 代码解读

无论你怎么写，都没办法在Spark 1.6 和 Spark 2.0 同时表现正常，总是会报错的，因为 Vector,Vectors等类的包名都发生了变化。

在Spark中，你可以通过 org.apache.spark.SPARK_VERSION 获取Spark的版本。我们定义一个类：

object SparkCompatibility {  
def sparkVersion = {     org.apache.spark.SPARK_VERSION  }
        
          
        
        
        
          
          AI 代码解读

动态编译

Scala 是可以很方便的动态源码的，对于而且代码也很简单，大体如下：

object ScalaSourceCodeCompiler {

  def compileCode(code: String): Any = {
    import scala.reflect.runtime.universe._
    val cm = runtimeMirror(Utils.getContextOrSparkClassLoader)
    val toolbox = cm.mkToolBox()
    val tree = toolbox.parse(code)
    val ref = toolbox.compile(tree)()
    ref
  }
        
          
        
        
        
          
          AI 代码解读

这种效果和Spark Shell里是类似的，ref 是里面的一个返回值。正常情况，你可以写两份代码，一份Spark 1.6. 一份Spark 2.0 ,然后在运行时决定编译那份代码。然而这种方式有一个缺点，尤其是在Spark中很难避免，如果compileCode 返回的值ref是需要被序列化到Executor的，则反序列化会导致问题，因为里面生成的一些匿名类在Executor中并不存在。除此之外，这种方法是实现兼容最有效的办法。

原先我考虑过使用Spark 内部的CodeGen代码，大致如下：

    def compileCode3[T](codeBody: String, references: Array[Any]): T = {
      val code = CodeFormatter.stripOverlappingComments(
        new CodeAndComment(codeBody, Map()))

      val c = CodeGenerator.compile(code)
      c.generate(references).asInstanceOf[T]
    }
        
          
        
        
        
          
          AI 代码解读

不幸的是，这个API本身也在变化，譬如CodeAndComment就是只有2.0才有的。

分离项目方式

将API有变化的部分，独立出来。比如前面我们提到的，对于Vector相关的API,1.6 和2.0 是不同的，那么我们可以剥离出两个工程，每个工程适配对应的版本，然后发布jar包，在Maven中根据Profile机制，根据Spark版本引入不同的适配包。这种方式相对来说比较繁琐。

反射的方法(StreamingPro在使用的方式)

下面是我用来解决Vector 包名变化的代码：

def vectorizeByReflect(vectorSize: Int) = {

    val clzzName =
      if (org.apache.spark.SPARK_VERSION.startsWith("2")) {
        "org.apache.spark.ml.linalg.Vectors"
      } else {
        "org.apache.spark.mllib.linalg.Vectors"
      }

    val reslutClzzName = if (org.apache.spark.SPARK_VERSION.startsWith("2")) {
      "org.apache.spark.ml.linalg.Vector"
    } else {
      "org.apache.spark.mllib.linalg.Vector"
    }
    def dense(v: Array[Double]) = {
      Class.forName(clzzName).getMethod("dense", classOf[Array[Double]]).invoke(null, v)
    }

    def sparse(vectorSize: Int, v: Array[(Int, Double)]) = {
      val method = Class.forName(clzzName).getMethod("sparse", classOf[Int], classOf[Seq[(Int, Double)]])
      val vs: Integer = vectorSize
      method.invoke(null, vs, v.toSeq)
    }

    val t = functions2.udf(reslutClzzName, (features: String) => {
      if (!features.contains(":")) {
        val v = features.split(",|\\s+").map(_.toDouble)
        dense(v)
      } else {
        val v = features.split(",|\\s+").map(_.split(":")).map(f => (f(0).toInt, f(1).toDouble))
        sparse(vectorSize, v)
      }
    })

    t
  }
        
          
        
        
        
          
          AI 代码解读

我们根据不同版本，动态加载对应的类，然后通过反射来调用方法，从而避免编译时错误。然而通过反射，就无法使用类似的代码了：

val t = udf {
 .....
}
        
          
        
        
        
          
          AI 代码解读

因为 udf 函数要求能够推导出输入和返回值是什么。而如果通过反射，因为返回值我们无法确定(有可能是org.apache.spark.ml.linalg.Vector，也有可能是org.apache.spark.mllib.linalg.Vector)，这个时候就无法通过编译了。于是我们改写了udf的是实现，然而这个实现也遇到了挫折，因为里面用到比如UserDefinedFunction类，已经在不同的包里面了，我们依然通过放射的方案解决：

def udf[RT: TypeTag, A1: TypeTag](className: String, f: Function1[A1, RT]): Any = {

    if (SparkCompatibility.sparkVersion.startsWith("2")) {
      val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType :: Nil).toOption
      val dufReg = Class.forName("org.apache.spark.sql.types.UDTRegistration").
        getMethod("getUDTFor", classOf[String]).invoke(null, className).asInstanceOf[Option[Class[_]]]
      val udt = dufReg.get.newInstance().asInstanceOf[UserDefinedType[_]]
      Class.forName("org.apache.spark.sql.expressions.UserDefinedFunction").
        getConstructor(classOf[AnyRef], classOf[DataType], classOf[Option[Seq[DataType]]]).
        newInstance(f, Schema(udt, nullable = true).dataType, inputTypes)
    } else {
      val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType :: Nil).toOption.get.toSeq
      val udt = org.apache.spark.util.Utils.classForName(className)
        .getAnnotation(classOf[SQLUserDefinedType]).udt().newInstance().asInstanceOf[UserDefinedType[_]]
      Class.forName("org.apache.spark.sql.UserDefinedFunction").
        getConstructor(classOf[AnyRef], classOf[DataType], classOf[Seq[DataType]]).
        newInstance(f, Schema(udt, nullable = true).dataType, inputTypes)

    }
  }
        
          
        
        
        
          
          AI 代码解读

很丑陋，对不对。这里还有一个问题，虽然udf返回的都是UserDefinedFunction对象，然而他们也是版本不兼容的，也就是我们无法让编译器确定返回的值是什么。我们使用了另外一个Scala语法的技巧，如下：

val t = functions2.udf(reslutClzzName, (features: String) => {
      if (!features.contains(":")) {
        val v = features.split(",|\\s+").map(_.toDouble)
        dense(v)
      } else {
        val v = features.split(",|\\s+").map(_.split(":")).map(f => (f(0).toInt, f(1).toDouble))
        sparse(vectorSize, v)
      }
    }).asInstanceOf[{def apply(exprs: Column*): Column}]
        
          
        
        
        
          
          AI 代码解读

核心在最后一行，我们声称返回的对象满足这个签名：

{def apply(exprs: Column*): Column}
        
          
        
        
        
          
          AI 代码解读

这个时候，就可以直接使用了：

training.select(  
col("label") cast (org.apache.spark.sql.types.DoubleType),  
t(col("features")) as "features")

        
          
        
        
        
          
          AI 代码解读