+--------+----+-----+

如同ProtocolBuffer，Avro，Thrift一样，Parquet也是支持元数据合并的。用户可以在一开始就定义一个简单的元数据，然后随着业务需要，逐渐往元数据中添加更多的列。在这种情况下，用户可能会创建多个Parquet文件，有着多个不同的但是却互相兼容的元数据。Parquet数据源支持自动推断出这种情况，并且进行多个Parquet文件的元数据的合并。

因为元数据合并是一种相对耗时的操作，而且在大多数情况下不是一种必要的特性，从Spark 1.5.0版本开始，默认是关闭Parquet文件的自动合并元数据的特性的。可以通过以下两种方式开启Parquet数据源的自动合并元数据的特性：
1、读取Parquet文件时，将数据源的选项，mergeSchema，设置为true
2、使用SQLContext.setConf()方法，将spark.sql.parquet.mergeSchema参数设置为true

scala代码：

package com.spark.study.core

import org.apache.spark.sql.SQLContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SaveMode

object parquettest3 {
def main(args:Array[String]){
    val conf = new SparkConf()
                  .setAppName("loadandsave")

    val sc = new SparkContext(conf)
    val sqlcontext = new SQLContext(sc)
    import sqlcontext.implicits._
    val info1 = Array(("zhangsan",23),("lisi",25)).toSeq
    val df1 = sc.parallelize(info1, 3).toDF("name","age")
    df1.save("hdfs://master:9000/students","parquet",SaveMode.Append)

    val info2 = Array(("wangwu","A"),("liu","B")).toSeq
    val df2 = sc.parallelize(info2, 3).toDF("name","grade")
    df2.save("hdfs://master:9000/students","parquet",SaveMode.Append)
    val res = sqlcontext.read.option("mergeSchema","true").parquet("hdfs://master:9000/students")
    res.printSchema()
    res.show()


}
}

ITPub博客

spark系列二：parquet数据源之自动合并元数据

最新文章