NotSerializableException

class Upserter(internalURL: String, internalUser: String, internalPassword: String) extends Serializable{
  lazy val internalConnection = DriverManager.getConnection(internalURL, internalUser, internalPassword)
  def upsert(primarykey: String, bookmark: String, db: String, table: String) = {
	val statement = internalConnection.createStatement()
	val upsertQuery = "INSERT INTO ..."
	val resultSet = statement.executeQuery(upsertQuery)
	internalConnection.close()
  }
}

val PKsAffectedDF: DataFrame = ...
val upserter = new Upserter(internalURL, internalUser, internalPassword)
PKsAffectedDF.map { x => upserter.upsert(x(0).toString(),x(1).toString(),dataSource.db,dataSource.table) }