How to insert few columns in a Table using spark

Klaus 0
Viewed 32

I am getting some fields from Multiple tables using joins using a select statement(5 fields). enter image description here

I have to insert these values in another Table Table-B which is having more columns(10 columns). enter image description here

How to insert these values in the Table_B like Col1 to id column and Col2 in Alias and Col3 in emp_age and Col4 in occupation.

I am getting the first result of multiple joins in a dataset.

Dataset exlCompaniesDataset = sparkSession.sql("Select query with multiple inner joins");

How to get each column values from Dataset and insert it in Table-B ?

bigdata java apache-spark apache-spark-sql
1 Answers

You essentially need to solve 2 issues -

  1. Make the initial dataframe to have the same number of columns as your target
  2. Rename the columns as needed from source to target.

Below is a sample code to do this. I have shown the spark-shell output at places to make it easier to understand.

import scala.util.Try
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

//Input Data
val inputDF = spark.sql("SELECT '1' as Col1,'name' as Col2,'age' as Col3,'job' as Col4")
scala> inputDF.show(false)
+----+----+----+----+
|Col1|Col2|Col3|Col4|
+----+----+----+----+
|1   |name|age |job |
+----+----+----+----+

//make same number of columns as needed in target
val inputColList = List("Col1","Col2","Col3","Col4","Col5","Col6","Col7","Col8","Col9","Col10")

var newDf = inputDF
//Loop and add any missing columns as Null
inputColList.foreach( fieldName => {
  if(!Try(newDf(fieldName)).isSuccess)
  newDf = newDf.withColumn(fieldName, lit(null).cast(StringType))
})

scala> newDf.show(false)
+----+----+----+----+----+----+----+----+----+-----+
|Col1|Col2|Col3|Col4|Col5|Col6|Col7|Col8|Col9|Col10|
+----+----+----+----+----+----+----+----+----+-----+
|1   |name|age |job |null|null|null|null|null|null |
+----+----+----+----+----+----+----+----+----+-----+

//Create a Source to Target Map.
val srcTgtMap = Map ("Col1"->"id","Col2"->"Alias","Col3"->"ten_id","Col4"->"occupation","Col5"->"emp__age","Col6"->"Salary","Col7"->"created_date","Col8"->"Modified_date","Col9"->"some_col1","Col10"->"col_col2" )

//Iterate to get new column names.

val colMappingList = srcTgtMap.keys.map(key => col(key).as(srcTgtMap(key))).toList
val dfRenamed = newDf.select(colMappingList: _*)

scala> dfRenamed.show(false)
+--------+--------+------+-------------+------+------------+----------+---------+---+-----+
|col_col2|emp__age|ten_id|Modified_date|Salary|created_date|occupation|some_col1|id |Alias|
+--------+--------+------+-------------+------+------------+----------+---------+---+-----+
|null    |null    |age   |null         |null  |null        |job       |null     |1  |name |
+--------+--------+------+-------------+------+------------+----------+---------+---+-----+

Hope that helps!!

Amit Sethi0
Related