• 售前

  • 售后

热门帖子
入门百科

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

[复制链接]
荷叶224 显示全部楼层 发表于 2021-10-26 12:27:55 |阅读模式 打印 上一主题 下一主题
dataframe是在spark1.3.0中推出的新的api,这让spark具备了处置惩罚大规模结构化数据的本领,在比原有的RDD转化方式易用的条件下,据说盘算性能更还快了两倍。spark在离线批处置惩罚大概实时盘算中都可以将rdd转成dataframe进而通过简朴的sql下令对数据进行操纵,对于熟悉sql的人来说在转换和过滤过程很方便,甚至可以有更高条理的应用,好比在实时这一块,传入kafka的topic名称和sql语句,背景读取本身设置好的内容字段反射成一个class并使用收支的sql对实时数据进行盘算,这种情况下不会spark streaming的人也都可以方便的享受到实时盘算带来的利益。    
下面的示例为读取本地文件成rdd并隐式转换成dataframe对数据进行查询,末了以追加的情势写入mysql表的过程,scala代码示比方下
  1. import java.sql.Timestamp
  2. import org.apache.spark.sql.{SaveMode, SQLContext}
  3. import org.apache.spark.{SparkContext, SparkConf}
  4. object DataFrameSql {
  5. case class memberbase(data_date:Long,memberid:String,createtime:Timestamp,sp:Int)extends Serializable{
  6. override def toString: String="%d\t%s\t%s\t%d".format(data_date,memberid,createtime,sp)
  7. }
  8. def main(args:Array[String]): Unit ={
  9. val conf = new SparkConf()
  10. conf.setMaster("local[2]")
  11. // ----------------------
  12. //参数 spark.sql.autoBroadcastJoinThreshold 设置某个表是否应该做broadcast,默认10M,设置为-1表示禁用
  13. //spark.sql.codegen 是否预编译sql成java字节码,长时间或频繁的sql有优化效果
  14. // spark.sql.inMemoryColumnarStorage.batchSize 一次处理的row数量,小心oom
  15. //spark.sql.inMemoryColumnarStorage.compressed 设置内存中的列存储是否需要压缩
  16. // ----------------------
  17. conf.set("spark.sql.shuffle.partitions","20") //默认partition是200个
  18. conf.setAppName("dataframe test")
  19. val sc = new SparkContext(conf)
  20. val sqc = new SQLContext(sc)
  21. val ac = sc.accumulator(0,"fail nums")
  22. val file = sc.textFile("src\\main\\resources\\000000_0")
  23. val log = file.map(lines => lines.split(" ")).filter(line =>
  24.   if (line.length != 4) { //做一个简单的过滤
  25.   ac.add(1)
  26.   false
  27.   } else true)
  28.   .map(line => memberbase(line(0).toLong, line(1),Timestamp.valueOf(line(2)), line(3).toInt))
  29. // 方法一、利用隐式转换
  30. import sqc.implicits._
  31. val dftemp = log.toDF() // 转换
  32. /*
  33.   方法二、利用createDataFrame方法,内部利用反射获取字段及其类型
  34.   val dftemp = sqc.createDataFrame(log)
  35.   */
  36. val df = dftemp.registerTempTable("memberbaseinfo")
  37. /*val sqlcommand ="select date_format(createtime,'yyyy-MM')as mm,count(1) as nums " +
  38.   "from memberbaseinfo group by date_format(createtime,'yyyy-MM') " +
  39.   "order by nums desc,mm asc "*/
  40. val sqlcommand="select * from memberbaseinfo"
  41. val sel = sqc.sql(sqlcommand)
  42. val prop = new java.util.Properties
  43. prop.setProperty("user","etl")
  44. prop.setProperty("password","xxx")
  45. // 调用DataFrameWriter将数据写入mysql
  46. val dataResult = sqc.sql(sqlcommand).write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/test","t_spark_dataframe_test",prop) // 表可以不存在
  47. println(ac.name.get+" "+ac.value)
  48. sc.stop()
  49. }
  50. }
复制代码
上面代码textFile中的示例数据如下,数据来自hive,字段信息分别为 分区号、用户id、注册时间、第三方号
  1. 20160309 45386477 2012-06-12 20:13:15 901438
  2. 20160309 45390977 2012-06-12 22:38:06 901036
  3. 20160309 45446677 2012-06-14 21:57:39 901438
  4. 20160309 45464977 2012-06-15 13:42:55 901438
  5. 20160309 45572377 2012-06-18 14:55:03 902606
  6. 20160309 45620577 2012-06-20 00:21:09 902606
  7. 20160309 45628377 2012-06-20 10:48:05 901181
  8. 20160309 45628877 2012-06-20 11:10:15 902606
  9. 20160309 45667777 2012-06-21 18:58:34 902524
  10. 20160309 45680177 2012-06-22 01:49:55
  11. 20160309 45687077 2012-06-22 11:23:22 902607
复制代码
这里留意字段类型映射,即case class类到dataframe映射,从官网的截图如下

更多明细可以检察官方文档 Spark SQL and DataFrame Guide
以上这篇spark rdd转dataframe 写入mysql的实例解说就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持脚本之家。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有帐号?立即注册

x

帖子地址: 

回复

使用道具 举报

分享
推广
火星云矿 | 预约S19Pro,享500抵1000!
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

草根技术分享(草根吧)是全球知名中文IT技术交流平台,创建于2021年,包含原创博客、精品问答、职业培训、技术社区、资源下载等产品服务,提供原创、优质、完整内容的专业IT技术开发社区。
  • 官方手机版

  • 微信公众号

  • 商务合作