spark 2.3.1 insertinto remove value of fields and change it to null

Question

I just upgrade my spark cluster from 2.2.1 to 2.3.1 in order to enjoy the feature of overwrite specific partitions. see link.

But .... From some reason when I am testing it I get a very strange behavior see code:

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
case class MyRow(partitionField: Int, someId: Int, someText: String)
object ExampleForStack2 extends App{
  val sparkConf = new SparkConf()
  sparkConf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
  sparkConf.setMaster(s"local[2]")
  val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  val list1 = List(
    MyRow(1, 1, "someText")
      ,MyRow(2, 2, "someText2")
  )
  val list2 = List(
    MyRow(1, 1, "someText modified")
    ,MyRow(3, 3, "someText3")
  )
  val df = spark.createDataFrame(list1)
  val df2 = spark.createDataFrame(list2)

  df2.show(false)
  df.write.partitionBy("partitionField").option("path","/tmp/tables/").saveAsTable("my_table")
  df2.write.mode(SaveMode.Overwrite).insertInto("my_table")
  spark.sql("select * from my_table").show(false)
}

And output:

+--------------+------+-----------------+
|partitionField|someId|someText         |
+--------------+------+-----------------+
|1             |1     |someText modified|
|3             |3     |someText3        |
+--------------+------+-----------------+

+------+---------+--------------+
|someId|someText |partitionField|
+------+---------+--------------+
|2     |someText2|2             |
|1     |someText |1             |
|3     |3        |null          |
|1     |1        |null          |
+------+---------+--------------+

Why I get those nulls ? It seems that fields were moved ? but why?

Thanks

score 0 · Answer 1 · answered Jan 02 '19 at 21:51

Ok found it, insert into is based on fields position. see documentation

Unlike saveAsTable, insertInto ignores the column names and just uses position-based resolution. For example:

scala> Seq((1, 2)).toDF("i", "j").write.mode("overwrite").saveAsTable("t1")
scala> Seq((3, 4)).toDF("j", "i").write.insertInto("t1")
scala> Seq((5, 6)).toDF("a", "b").write.insertInto("t1")
scala> sql("select * from t1").show
+---+---+
|  i|  j|
+---+---+
|  5|  6|
|  3|  4|
|  1|  2|
+---+---+

Because it inserts data to an existing table, format or options will be ignored.

Moreover I am using dynamic partition which should appear as the last field. So the solution is to move the dynamic partitions to the end of the dataframe, which means in my case:

df2.select("someId", "someText","partitionField").write.mode(SaveMode.Overwrite).insertInto("my_table")

spark 2.3.1 insertinto remove value of fields and change it to null

1 Answers1