博客

spark sql编程之实现合并Parquet格式的DataFrame的schema

  • Post author:
  • Post category:其他



问题导读







1.DataFrame合并schema由哪个配置项控制？






2.修改配置项的方式有哪两种？






3.spark读取hive parquet格式的表，是否转换为自己的格式？







首先说下什么是schema，其实这跟通俗来讲，与我们传统数据表字段的名称是一个意思。明白了这个，我们在继续往下看。






合并schema




首先创建RDD，并转换为含有两个字段”value”, “square”的DataFrame


[Scala]

纯文本查看


复制代码


?

1

val


squaresDF


=


spark.sparkContext.makeRDD(


1


to


5


).map(i


=


> (i, i * i)).toDF(


"value"


,


"square"


)









然后以parquet格式保存


[Scala]

纯文本查看


复制代码


?

1

squaresDF.write.parquet(


"data/test_table/key=1"


)






然后在创建RDD，并转换为含有两个字段”value”, “cube”的DataFrame


[Scala]

纯文本查看


复制代码


?

1

val


cubesDF


=


spark.sparkContext.makeRDD(


6


to


10


).map(i


=


> (i, i * i * i)).toDF(


"value"


,


"cube"


)











然后以parquet格式保存




[Scala]

纯文本查看


复制代码


?

1

cubesDF.write.parquet(


"data/test_table/key=2"


)




最后合并schema


[Scala]

纯文本查看


复制代码


?

1

val


mergedDF


=


spark.read.option(


"mergeSchema"


,


"true"


).parquet(


"data/test_table"


)











我们打印schema


[Scala]

纯文本查看


复制代码


?

1

mergedDF.printSchema()











接着我们现实数据


[Scala]

纯文本查看


复制代码


?

1

mergedDF.show











如果想合并schema需要设置mergeSchema 为true，当然还有另外一种方式是设置spark.sql.parquet.mergeSchema为true。






相关补充说明：






Hive metastore Parquet表格式转换




当读取hive的 Parquet 表时，Spark SQL为了提高性能，会使用自己的支持的Parquet，由配置 spark.sql.hive.convertMetastoreParquet控制，默认是开启的。




上面除了Parquet格式支持外，还有ProtocolBuffer, Avro, 和Thrift支持合并。






如何修改配置项：




可以通过SparkSession 的setConf 或则使用SQL命令


[Bash shell]

纯文本查看


复制代码


?

1

SET key=value




更多配置项如下：