spark rdd转dataframe 写入mysql的实例讲解

import java.sql.Timestamp
import org.apache.spark.sql.{SaveMode, SQLContext}
import org.apache.spark.{SparkContext, SparkConf}
object DataFrameSql {
 case class memberbase(data_date:Long,memberid:String,createtime:Timestamp,sp:Int)extends Serializable{
 override def toString: String="%d\t%s\t%s\t%d".format(data_date,memberid,createtime,sp)
 }
 def main(args:Array[String]): Unit ={
 val conf = new SparkConf()
 conf.setMaster("local[2]")
// ----------------------
 //参数 spark.sql.autoBroadcastJoinThreshold 设置某个表是否应该做broadcast，默认10M，设置为-1表示禁用
 //spark.sql.codegen 是否预编译sql成java字节码，长时间或频繁的sql有优化效果
 // spark.sql.inMemoryColumnarStorage.batchSize 一次处理的row数量，小心oom
 //spark.sql.inMemoryColumnarStorage.compressed 设置内存中的列存储是否需要压缩
// ----------------------
 conf.set("spark.sql.shuffle.partitions","20") //默认partition是200个
 conf.setAppName("dataframe test")
 val sc = new SparkContext(conf)
 val sqc = new SQLContext(sc)
 val ac = sc.accumulator(0,"fail nums")
 val file = sc.textFile("src\\main\\resources\\000000_0")
 val log = file.map(lines => lines.split(" ")).filter(line =>
  if (line.length != 4) { //做一个简单的过滤
  ac.add(1)
  false
  } else true)
  .map(line => memberbase(line(0).toLong, line(1),Timestamp.valueOf(line(2)), line(3).toInt))
 // 方法一、利用隐式转换
 import sqc.implicits._
 val dftemp = log.toDF() // 转换
 
 val df = dftemp.registerTempTable("memberbaseinfo")
 
 val sqlcommand="select * from memberbaseinfo"
 val sel = sqc.sql(sqlcommand)
 val prop = new java.util.Properties
 prop.setProperty("user","etl")
 prop.setProperty("password","xxx")
 // 调用DataFrameWriter将数据写入mysql
 val dataResult = sqc.sql(sqlcommand).write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/test","t_spark_dataframe_test",prop) // 表可以不存在
 println(ac.name.get+" "+ac.value)
 sc.stop()
 }
}