spark sql與mysql數據怎么加載并導出數據到mysql

本篇內容主要講解“spark sql與mysql數據怎么加載并導出數據到mysql”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“spark sql與mysql數據怎么加載并導出數據到mysql”吧!

package hgs.spark.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.SQLImplicits
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.Row
import java.util.Properties
//第一種方法創建dataframe
object SqlTest1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("sqltest1").setMaster("local")
    val context = new SparkContext(conf)
    val sqlContext = new SQLContext(context)
    
    val rdd = context.textFile("d:\\person",1)
    val rdd2 = rdd.map(x=>{val t = x.split(" ");person(t(0).toInt,t(1),t(2).toInt)})
    val rdd3 = rdd.map(x=>{val t = x.split(" ");person(t(0).toInt,t(1),t(2).toInt)})
    //第一種方法創建dataframe,在這里需要導入隱式轉換
    import sqlContext.implicits._  
    
    val persondf = rdd2.toDF() 
    
    val persondf3 = rdd3.toDF()
    
   
    //這個方法在2.1.0里面被廢除
    //persondf.registerTempTable("person")
    //使用該函數代替
    persondf.createOrReplaceTempView("person")
    persondf3.createOrReplaceTempView("person2")
    val result = sqlContext.sql("select a.id ,b.name  from person a ,person2 b   where a.id = b.id")
    //打印查詢的結果
    result.show()
    //或者將結果保存到文件
    //result.write.json("d://personselect")
    val props =  new Properties();
    props.put("user", "root")
    props.put("password", "123456")
 
    result.write.mode("append").jdbc("jdbc:mysql://192.168.6.133:3306/hgs", "person", props)
   //result.write.mode(saveMode)
    context.stop()
  }
}
case class person(id:Int,name:String,age:Int)
object sqlTest3{
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("sqltest3").setMaster("local")
    val context = new SparkContext(conf)
    val sqlContext = new SQLContext(context)
    val persontable = sqlContext.read.format("jdbc").option("url", "jdbc:mysql://192.168.6.133:3306/hgs")
                                  .option("dbtable", "person")
                                  .option("user", "root")
                                  .option("password", "123456").load().createOrReplaceTempView("person")
                
                                  
    sqlContext.sql("select id,concat(name,\"xxx\") from person").show()
    context.stop()
  }
}

到此，相信大家對“spark sql與mysql數據怎么加載并導出數據到mysql”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！