spark创建空dataframe

Posted 2023-03-24 bitcarmanlee

1.emptyDataFrame

项目中会需要用到创建空的dataframe。spark提供了emptyDataFrame方法，可以直接创建。

  /**
   * Returns a `DataFrame` with no rows or columns.
   *
   * @since 2.0.0
   */
  @transient
  lazy val emptyDataFrame: DataFrame = 
    createDataFrame(sparkContext.emptyRDD[Row], StructType(Nil))
  

源码中已经很明确告诉我们，emptyDataFrame创建了一个不含任何行列且schema为空的dataframe。

这个空的dataframe在实际中用途有限，比如如下场景。

	def t2() = 
		val sparkConf = new SparkConf().setMaster("local[2]")
		val spark = SparkSession.builder().config(sparkConf).getOrCreate()
		import spark.implicits._
		
		
		val seq = Seq(("id1", 1), ("id1", 1), ("id1", 2), ("id2", 1), ("id2", 1))
		val df = spark.sparkContext.parallelize(seq, 1).toDF("id", "num")
		var emptydf = spark.emptyDataFrame
		
		emptydf = emptydf.union(df)
		emptydf.show(10)
	

上面的方法运行时候直接抛出错误信息

org.apache.spark.sql.AnalysisException: Union can only be performed on tables with the same number of columns, but the first table has 0 columns and the second table has 2 columns;;
'Union
:- LogicalRDD
+- Project [_1#3 AS id#6, _2#4 AS num#7]
   +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple2, true], top level Product input object)._1, true) AS _1#3, assertnotnull(input[0, scala.Tuple2, true], top level Product input object)._2 AS _2#4]
      +- ExternalRDD [obj#2]

错误信息也很直观，emptydf是0 columns，df是2 columns，所以无法直接union。

2.emptyDataFrame指定schema

为了解决上面的问题，我们需要在生成emptyDataFrame的时候指定schema。

import org.apache.spark.sql.types.IntegerType, StringType, StructField, StructType

	def t2() = 
		val sparkConf = new SparkConf().setMaster("local[2]")
		val spark = SparkSession.builder().config(sparkConf).getOrCreate()
		import spark.implicits._
		
		
		val seq = Seq(("id1", 1), ("id1", 1), ("id1", 2), ("id2", 1), ("id2", 1))
		val df = spark.sparkContext.parallelize(seq, 1).toDF("id", "num")
		
		val schema = StructType(
			Seq(StructField("id", StringType, true),
				StructField("num", IntegerType, true))
		)
		var emptydf = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)
		
		emptydf = emptydf.union(df)
		emptydf.show(10)
	

通过StructType与StructField配合使用，就可以实现上述目的。