13518219792
本篇内容主要讲解“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”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
