JUNIT Test Cases- Spark JDBC

Viewed 685

I am new to spark with Java programming. I have a method to read data from Oracle database. Now i need help in writing test cases using JUnit Framework for the below code.

Dataset df = spark.read().format("jdbc").jdbc(jdbcUrl, dbTable1, connectionProperties);

1 Answers

You can using the below approach, You need to use H2 library which will create an in memory temporary server with database, in which you can create the required table with test data. this example is using MS Sql Server (MODE=MSSQLServer), you can change it.

<!-- https://mvnrepository.com/artifact/com.h2database/h2 -->
<dependency>
  <groupId>com.h2database</groupId>
  <artifactId>h2</artifactId>
  <version>1.4.200</version>
  <scope>test</scope>
</dependency>

Actual Method

def GetJDBCDataframe(spark: SparkSession, jdbcUrl: String, connectionProperties :Properties): DataFrame = {
    val employeesQuery = "(SELECT * FROM Employee) ref_alias"
    val df = spark.read.jdbc(url = jdbcUrl, table = employeesQuery, connectionProperties)
    df
}

Testcase method:

@Test
def GetJDBCDataframe(): Unit = {

  val spark: SparkSession = {
    SparkSession
      .builder()
      .master("local[2]")
      .appName("SampleSparkScalaTest")
      .getOrCreate()
  }
  //H2 library connection for creating a inmemory MSSQL databse
  val url = "jdbc:h2:mem:Mydatabasename;MODE=MSSQLServer"

  val connectionProperties = new Properties()
  connectionProperties.setProperty("user", "sa")
  connectionProperties.setProperty("password", "")

  Class.forName("org.h2.Driver")
  val conn = DriverManager.getConnection(url, connectionProperties)

  //create mock jdbc Employee table (immemory)
  conn.prepareStatement("create table Employee(EmpId int, FirstName varchar, LastName varchar)").executeUpdate()
  conn.prepareStatement("insert into Employee values (1, 'Brain', 'Lara')").executeUpdate()
  conn.prepareStatement("insert into Employee values (2, 'Virat', 'Kholi')").executeUpdate()
  conn.prepareStatement("insert into Employee values (3, 'MS', 'Dhoni')").executeUpdate()
  conn.commit()

  val rowCount = TransactionLogicTransformation.GetJDBCDataframe(spark,url,connectionProperties).count()

  assertEquals(3, rowCount)
}
Related