Блог

Я работаю с Spark 3.1.2 И. Scala 2.12 У меня есть функция чтения, которая возвращает resultset тип. Код находится в следующем:

 def read_data(group_id: Int):java.sql.ResultSet =  {
 val table_name = "TABLE"
 val col_name = "COL"
 val query =
  """ select ID,F1,F2,F3,F4,F5,F6,F7
    | from """.stripMargin   table_name   """ where MOD(TO_NUMBER(substr(""" col_name """, -LEAST(2, LENGTH(""" col_name """)))),""" num_node """)=""" group_id
  
 val oracleUser = "ORCL"
 val oraclePassword = "XXXXXX"
 val oracleURL = "jdbc:oracle:thin:@//x.x.x.x:1521/ORCLDB"

 val ods = new OracleDataSource()
 ods.setUser(oracleUser)
 ods.setURL(oracleURL)
 ods.setPassword(oraclePassword)

 val con = ods.getConnection()
 val statement = con.createStatement()

 statement.setFetchSize(1000)     
 println("statement:",statement)

 val  resultSet : java.sql.ResultSet = statement.executeQuery(query) //
 return resultSet
  }

 val conf = new SparkConf()
  .setMaster("local[2]")
  .setAppName("testScala")
  .set("spark.executor.memory", "8g")
  .set("spark.executor.cores", "2")
  .set("spark.task.cpus","1")
 val sc = new SparkContext(conf)
 val rdd = sc.parallelize(group_list,num_node)
      .map(read_data).collect()
 println("rdd:",rdd)
 

Когда я запускаю приведенный выше код, он показывает мне эту ошибку:

   ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
  java.io.NotSerializableException: oracle.jdbc.driver.ForwardOnlyResultSet
 

Не могли бы вы, пожалуйста, подсказать мне, как сделать resultSet сериализуемым?

Любая помощь действительно ценится.