Spark Streaming | Write different data frames to multiple tables in Synapse DW

user_15559764 0
Viewed 931

I have multiple data frames which are extracted from one json messages coming in azure-event-hub. We would like to push these DFs to separate tables in Synapse DW using spark streaming job.

here is my schema -

root
 |-- Name: string (nullable = true)
 |-- Salary: string (nullable = true)
 |-- EmpID: string (nullable = true)
 |-- Projects: struct (nullable = true)
 |    |-- ProjectID: string (nullable = true)
 |    |-- ProjectName: string (nullable = true)
 |    |-- Duration: string (nullable = true)
 |    |-- Location: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- City: string (nullable = true)
 |    |    |    |-- State: string (nullable = true)
 |    |-- Contact: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- Phone: string (nullable = true)
 |    |    |    |-- email: string (nullable = true)

I have extracted 4 different data frames from above schema -

  1. Project
  2. Location
  3. Contact
  4. Employee

They all should be inserted in 4 different tables in Synapse

ProjectDf.write.format("spark.sqldw").options(.dbo.Project).save(...)
LocationDf.write.format("spark.sqldw").options(.dbo.Loc).save(...)
ContactDf.write.format("spark.sqldw").options(.dbo.Contact).save(...)
EmployeeDf.write.format("spark.sqldw").options(.dbo.Emp).save(...)

Pls suggest how to apply ForeachBatch sink on this to insert tables.

spark-streaming spark-structured-streaming azure-databricks apache-spark scala
1 Answers

If you plan to write four different Dataframes based on a single input streaming Dataframe, you can make use of foreachBatch in the following way:

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>

  // as you plan to use the batchDF to create multiple output it might be wort persisting the batchDF
  batchDF.persist()

  // create the four different Dataframes based on the input
  val ProjectDf = batchDF.select(...)
  val LocationDf = batchDF.select(...) 
  val ContactDf = batchDF.select(...)
  val EmployeeDf = batchDF.select(...)

  // then you can save those four Dataframes into the desired locations
  ProjectDf.write.format("spark.sqldw").options(.dbo.Project).save(...)
  LocationDf.write.format("spark.sqldw").options(.dbo.Loc).save(...)
  ContactDf.write.format("spark.sqldw").options(.dbo.Contact).save(...)
  EmployeeDf.write.format("spark.sqldw").options(.dbo.Emp).save(...)

  // do not forget to unpersist your batchDF
  batchDF.unpersist()
}

This is described in the documentation on Using foreach and foreachBatch

If you are hitting the Exception "Overloaded method foreachBatch with alternatives" you could have a look at the release notes of the Databricks Runtime 7.0 where it says:

"To fix the compilation error, change foreachBatch { (df, id) => myFunc(df, id) } to foreachBatch(myFunc _) or use the Java API explicitly: foreachBatch(new VoidFunction2 ...)."

That means, your code would look like:

def myFunc(batchDF: DataFrame, batchId: Long): Unit = {
  // as you plan to use the batchDF to create multiple output it might be wort persisting the batchDF
  batchDF.persist()

  // create the four different Dataframes based on the input
  val ProjectDf = batchDF.select(...)
  val LocationDf = batchDF.select(...) 
  val ContactDf = batchDF.select(...)
  val EmployeeDf = batchDF.select(...)

  // then you can save those four Dataframes into the desired locations
  ProjectDf.write.format("spark.sqldw").options(.dbo.Project).save(...)
  LocationDf.write.format("spark.sqldw").options(.dbo.Loc).save(...)
  ContactDf.write.format("spark.sqldw").options(.dbo.Contact).save(...)
  EmployeeDf.write.format("spark.sqldw").options(.dbo.Emp).save(...)

  // do not forget to unpersist your batchDF
  batchDF.unpersist()
}


streamingDF.writeStream.foreachBatch(myFunc _).[...].start()
user_122089100
Related