Spark

Spark踩坑3-共享Scala集合类

最近为了更好的了解Spark的细节，改用Spark的亲儿子Scala进行开发，得益于框架API的统一，Spark方面没什么需要重新学习的，开发完就想着按照以前的经验优化一下。
记得官方Tuning Spark中，第一件要干的事就是调整序列化类库，Spark默认使用的Java serialization的序列化方式。官方建议使用Kryo类库进行序列化可以获得更好的压缩率和传输效率。
切换的方式很简单，只要在配置时指定序列化方式为:org.apache.spark.serializer.KryoSerializer即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
object Recommender {
  private val log = Logger(Recommender.getClass)

  // load config
  private val config = ConfigFactory.load("config.properties")

  // spark config
  private val appName = config.getString("appName")
  private val mode = config.getString("mode")


  def main(args: Array[String]): Unit = {
    log.info("init {}，mode:{}", appName, mode)

    // init spark
    val sc = new SparkConf()
      .setAppName(appName)
      .setMaster(mode)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")


    val session = SparkSession.builder().config(sc).getOrCreate()
    try {
      initMat(session)

      var headMap = new mutable.LinkedHashMap[String, Int]()
      headMap.put("ID", 0)
      headMap.put("UID", 1)

      val headBroadcast = ss.sparkContext.broadcast[mutable.LinkedHashMap[String, Int]](headMap)

      // 略...


    } finally {
      session.close()
    }
  }
}

但是当我切换序列化方式后，我发现我要广播的一个LinkedHashMap每次获取都是空的。
于是猜测Kryo框架是不是没办法序列化Scala的集合类型，因为大部分的序列化框架对集合类都有或多或少的一些限制，比如Protostuff等。
于是写个测试代码测试一下：

1
2
3
4
5
6
7
8
9
10
def main(args: Array[String]): Unit = {
    var headMap = new mutable.LinkedHashMap[String, Int]()
    headMap.put("a", 1)
    headMap.put("b", 2)
    headMap.put("c", 3)
    //
    val kryo = new Kryo()
    val output = new Output(new FileOutputStream("e:\\ser.bin"))
    kryo.writeObject(output, headMap)
  }

最终写到磁盘ser.bin的才1个字节，都不用反序列化，肯定不对。
查了一些官方文档，有个相对简单点的解决办法：

1
kryo.register(classOf[mutable.HashMap[String, (String, String)]], new JavaSerializer)

将原生类做为自定义的类注册下，并指定序列化方式仍然使用jdk的。。。
虽然问题能解决但这其实已经背离本意了，好在一点是mutable.HashMap使用jdk序列化，其他的类使用kryo，相比较以前全部用jdk效率确实高点。