spark生成大宽表的parquet性能优化

1.  背景介绍

　　将一份数据量很大的用户属性文件解析成结构化的数据供查询框架查询剖析，其中用户属性包含用户标识，平台类型，性别，年龄，学历，兴趣爱好，购物倾向等等，大概共有七百个左右的标签属性。为了查询框架能够快速查询出有特定标签的人群，将最终的存储结果定义为了将七百个左右的标签属性展平存储为parquet文件，这样每个标签属性对于用户而言只有存在和不存在两种情况。

2. 第一版实现过程

　　　第一步，将用户所有标签标识作为一个资源文件保存到spark中，并读取该资源文件的标签标识为一个标签集合(定义为listAll)，并通过sparkContext来进行广播；

　　   第二步，使用spark core读取hdfs上的用户属性文件(其中每行是一个用户所拥有的标签)，将单个用户所拥有的标签解析成一个标签集合(定义为listUser)，也就是说listUser是listAll的一个子集；

　　　第三步，对于单个用户而言，遍历步骤一的结果集listAll，对于每一个标签判断该用户是否存在，如果存在则将标签设置为1(表示存在)，否则设置为0(表示不存在)，并将所有标签及相应的值保存为一个Map(定义为map)

　　　第四步，根据第三步的map构造成spark sql中的Row

　　　第五步，依据第一步的集合listAll构造出spark sql的Schema

　　　第六步，将第四步和第五步的结果通过spark sql的createDataFrame构造成DataFrame。

　　　第七步，通过DataFrame.write.parquet(output)将结果保存到hdfs上

     通过上述的七步，认为已经很easy的处理完了这个需求，但是真正测试时发现性能比想象的要慢的多，严重的达不到性能要求。对于性能影响究竟出现在什么地方？初步猜测，问题出现在第四步，第六步，第七步的可能性比较大。经过实际的测试，发现性能主要消耗在第七步，其他步骤的执行都特别快。这样也就定位到了问题。

　　而且通过测试知道，生成parquet消耗的性能最高，生成json的话很快就能完成，如果不生成任何对象，而是直接foreach执行的话，性能会更高。而且相同数据量下，如果列数在七百多个时，json写入时间是parquet写入时间的三分之一，如果列数在四百个时，json写入时间是parquet写入时间的二分之一，如果列数在五十个，json写入时间是parquet写入时间的三分之二。也就是列数越少，json和parquet的写入速度越接近。至于为什么生成parquet性能很差，待后续分析spark sql的save方法。

　　测试的例子

  private def CTRL_A = '01'

  private def CTRL_B = '02'

  private def CTRL_C = '03'

  def main(args: Array[String]): Unit = {

    val resourcePath = this.getClass.getResource("/resource.txt").getFile
    val sourcePath = this.getClass.getResource("/*.gz").getFile
    val output = "/home/dev/output"

    val conf = new SparkConf().setAppName("user test").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sqlContext.setConf("spark.sql.parquet.binaryAsString", "true")
    sqlContext.setConf("spark.sql.inMemoryColumnarStorage.compressed", "true")
    sqlContext.setConf("spark.sql.parquet.compression.codec", "snappy")
    val map: Map[String, String] = buildResource(resourcePath)
    val schema = buildSchema(map)
    val bd = sc.broadcast(map)
    val bdSchema = sc.broadcast(schema)

    val start=System.currentTimeMillis()
    val rdd = sc.textFile(sourcePath)
      .map(line => {
        val map = buildUser(line, bd.value)
        buildRow(map._3, map._1, map._2)
      })
//    rdd.foreach(_=>())
//    sqlContext.createDataFrame(rdd, bdSchema.value).write.mode(SaveMode.Overwrite).json(output)
    sqlContext.createDataFrame(rdd, bdSchema.value).write.mode(SaveMode.Overwrite).parquet(output)
    val end = System.currentTimeMillis()
    System.out.print(end - start)
  }

  /**
    * 读取资源文件
    * @param file
    * @return
    */
  def buildResource(file: String): Map[String, String] = {
    val reader = Source.fromFile(file)
    val map = new mutable.HashMap[String, String]()
    for (line <- reader.getLines() if !Strings.isNullOrEmpty(line)) {
        val arr = StringUtils.splitPreserveAllTokens(line, '	')
        map.+=((arr(0), "0"))
    }

    map.toMap
  }

  /**
    * 生成用户属性
    * @param line
    * @param map
    * @return
    */
  def buildUser(line: String, map: Map[String, String]): (String, Int, Map[String, String]) = {
    if (Strings.isNullOrEmpty(line)) {
      return ("", 0, Map.empty)
    }
    val array = StringUtils.splitPreserveAllTokens(line, CTRL_A)
    val cookie = if (Strings.isNullOrEmpty(array(0))) "-" else array(0)
    val platform = array(1).toInt
    val base = buildFeature(array(2))
    val interest = buildFeature(array(3))
    val buy = buildFeature(array(4))
    val features = base ++ interest ++ buy
    val result = new mutable.HashMap[String, String]()
    for (pair <- map) {
      val value = if (features.contains(pair._1)) "1" else "0"
      result.+=((pair._1, value))
    }
    (cookie, platform, result.toMap)
  }

  /**
    * 抽取用户标签
    * @param expr
    * @return
    */
  def buildFeature(expr: String): Array[String] = {
    if (Strings.isNullOrEmpty(expr)) {
      return Array.empty
    }
    val arr = StringUtils.splitPreserveAllTokens(expr, CTRL_B)
    val buffer = new ArrayBuffer[String]()
    for (key <- arr) {
      val pair = StringUtils.splitPreserveAllTokens(key, CTRL_C)
      buffer += (s"_${pair(0)}")
    }
    buffer.toArray
  }

  /**
    * 动态生成DataFrame的Schema
    * @param map
    * @return
    */
  def buildSchema(map: Map[String, String]): StructType = {
    val buffer = new ArrayBuffer[StructField]()
    buffer += (StructField("user", StringType, false))
    buffer += (StructField("platform", IntegerType, false))
    for (pair <- map) {
      buffer += (StructField(s"_${pair._1}", IntegerType, true))
    }
    return StructType(List(buffer: _*))
  }

  /**
    * 将用户属性构造成Spark SQL的Row
    * @param map
    * @param user
    * @param platform
    * @return
    */
  def buildRow(map: Map[String, String], user: String, platform: Int): Row = {
    val buffer = new ArrayBuffer[Any]()
    buffer += (user)
    buffer += (platform)
    for (pair <- map) {
      buffer += (pair._2.toInt)
    }
    return Row(buffer: _*)
  }

3. 第二版实现过程

　　在第一版中初步怀疑是DataFrame在生成parquet时进行了一些特殊逻辑的处理，所以决定自己实现ParquetWriter方法来测试下性能，采用了avro来向parquet中写入数据。方法大概包含定义好avro资源文件，然后使用AvroParquetWriter类来向parquet中写入内容，具体的写入方法类似于https://blog.csdn.net/gg584741/article/details/51614752。通过这种方式来写入parquet，相同数据量的情况下，性能提升了一倍多。至于为什么性能有这么大的提升，有待后续研究。到此优化就告一段落了。

 　　在此优化期间，遇到了下列问题：

　　1.  avro 的资源文件在生成java类时，属性限制必须255个一下。该限制在https://issues.apache.org/jira/browse/AVRO-1642 提到。

      2.  java 类属性和方法参数也需要小于255个，详见https://docs.oracle.com/javase/specs/jvms/se7/html/jvms-4.html#jvms-4.11,https://stackoverflow.com/questions/30581531/maximum-number-of-parameters-in-java-method-declaration

 

     对于上述显示的解决方案是在maven配置文件中不适用avro-maven-plugin插件来自动生成java类，而是在程序运行时通过

val Schema = (new Schema.Parser()).parse(new File(file))

来动态生成Schema来供后续AvroParquetWriter使用。