spark streaming怎么将计算结果写入oracle

Posted 2023-02-19

参考技术A 你试试，看这样可以不！

object Test

case class Person(words: String,number:Int)
def main(args: Array[String])

val sparkConf = new SparkConf().setAppName("TextStream")
val sc = new SparkContext(sparkConf)
Class.forName("oracle.jdbc.driver.OracleDriver").newInstance()
val conn = DriverManager.getConnection("jdbc:oracle:thin:@host/oracledb", "user", "password")
var ps: PreparedStatement = null
val ssc = new StreamingContext(sparkConf, Seconds(6))
// Create the FileInputDStream on the directory and use the
val lines = ssc.textFileStream("/user/input/")
val words = lines.flatMap(_.split(","))
val wordCounts = words.map(x => (x, 1))
// wordCounts.print()
wordCounts.foreachRDD((rdd:RDD[(String,Int)])=>
rdd.foreach(lines=>
ps = conn.prepareStatement("insert into testtables values(?,?)")
ps.setString(1, lines._1)
ps.setInt(2, lines._2)
ps.executeUpdate()
)
)