我们知道Spark2.0 ,Spark 1.6还有Spark 1.5 三者之间版本是不兼容的,尤其是一些内部API变化比较大。如果你的系统使用了不少底层的API,那么这篇文章或许对你有帮助。我们介绍的兼容相关一些技巧,主要包括动态编译以及反射等方式,也用到了Scala的一些语言特性。
在Spark 1.6 时,大部分机器学习相关的类使用的向量还是
org.apache.spark.mllib.linalg.Vector
而到2.0后,已经基本都变更成
org.apache.spark.ml.linalg.Vector
同理对应的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" )
无论你怎么写,在Spark 1.6 或者Spark 2.0 里,总是会报错的。
在Spark中,你可以通过 org.apache.spark.SPARK_VERSION 获取Spark的版本。
我们定义一个类:
object SparkCompatibility { def sparkVersion = { org.apache.spark.SPARK_VERSION }
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 }
这种效果和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] }
不幸的是,这个API本身也在变化,譬如CodeAndComment就是只有2.0才有的。
下面是我用来解决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 }
我们根据不同版本,通过反射来调用对应的方法,从而避免编译时错误。然而通过反射,就无法使用类似的代码了:
val t = udf { ..... }
因为 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) } }
很丑陋,对不对。这里还有一个问题,虽然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}]
核心在最后一行,我们声称返回的对象满足这个签名:
{def apply(exprs: Column*): Column}
这个时候,就可以直接使用了:
training.select( col("label") cast (org.apache.spark.sql.types.DoubleType), t(col("features")) as "features")
Spark 1.6 和2.0 内部API变化很大,然而对普通用户的API兼容性还是不错的。做版本兼容似乎并不是一件容易的事情。所以当使用StreamingPro做机器学习相关工作时,我只兼容了Spark 1.6,2.0,而抛弃了 1.5版本。但是对于普通的ETL以及流式计算,三个版本都是支持的。