服务器之家:专注于服务器技术及软件下载分享
分类导航

Mysql|

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

2019-07-11 17:05Dino系我 Mysql

今天小编就为大家分享一篇spark rdd转dataframe 写入mysql的实例讲解,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

dataframe是在spark1.3.0中推出的新的api,这让spark具备了处理大规模结构化数据的能力,在比原有的RDD转化方式易用的前提下,据说计算性能更还快了两倍。spark在离线批处理或者实时计算中都可以将rdd转成dataframe进而通过简单的sql命令对数据进行操作,对于熟悉sql的人来说在转换和过滤过程很方便,甚至可以有更高层次的应用,比如在实时这一块,传入kafka的topic名称和sql语句,后台读取自己配置好的内容字段反射成一个class并利用出入的sql对实时数据进行计算,这种情况下不会spark streaming的人也都可以方便的享受到实时计算带来的好处。    

下面的示例为读取本地文件成rdd并隐式转换成dataframe对数据进行查询,最后以追加的形式写入mysql表的过程,scala代码示例如下

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() // 转换
 /*
  方法二、利用createDataFrame方法,内部利用反射获取字段及其类型
  val dftemp = sqc.createDataFrame(log)
  */
 val df = dftemp.registerTempTable("memberbaseinfo")
 /*val sqlcommand ="select date_format(createtime,'yyyy-MM')as mm,count(1) as nums " +
  "from memberbaseinfo group by date_format(createtime,'yyyy-MM') " +
  "order by nums desc,mm asc "*/
 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()
 }
}

上面代码textFile中的示例数据如下,数据来自hive,字段信息分别为 分区号、用户id、注册时间、第三方号

20160309 45386477 2012-06-12 20:13:15 901438
20160309 45390977 2012-06-12 22:38:06 901036
20160309 45446677 2012-06-14 21:57:39 901438
20160309 45464977 2012-06-15 13:42:55 901438
20160309 45572377 2012-06-18 14:55:03 902606
20160309 45620577 2012-06-20 00:21:09 902606
20160309 45628377 2012-06-20 10:48:05 901181
20160309 45628877 2012-06-20 11:10:15 902606
20160309 45667777 2012-06-21 18:58:34 902524
20160309 45680177 2012-06-22 01:49:55 
20160309 45687077 2012-06-22 11:23:22 902607

这里注意字段类型映射,即case class类到dataframe映射,从官网的截图如下

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

更多明细可以查看官方文档 Spark SQL and DataFrame Guide

以上这篇spark rdd转dataframe 写入mysql的实例讲解就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持服务器之家。

延伸 · 阅读

精彩推荐
  • MysqlMySQL limit性能分析与优化

    MySQL limit性能分析与优化

    今天小编就为大家分享一篇关于MySQL limit性能分析与优化,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧...

    wlmmmm4082019-06-07
  • Mysql解决Windows环境下安装 mysql-8.0.11-winx64 遇到的问题

    解决Windows环境下安装 mysql-8.0.11-winx64 遇到的问题

    这篇文章主要介绍了Windows环境下安装 mysql-8.0.11-winx64 遇到的问题及解决办法 ,需要的朋友可以参考下...

    color_小浣熊3962019-06-18
  • Mysql数据库中间件MyCat的介绍

    数据库中间件MyCat的介绍

    今天小编就为大家分享一篇关于数据库中间件MyCat的介绍,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧...

    qq_431937975272019-06-10
  • MysqlMySQL通过实例化对象参数查询实例讲解

    MySQL通过实例化对象参数查询实例讲解

    在本篇文章里我们给大家分享了关于MySQL如何通过实例化对象参数查询数据的相关知识点内容,有需要的朋友们可以测试参考下。...

    laozhang2162019-06-20
  • Mysqlmysql中插入emoji表情失败的原因与解决

    mysql中插入emoji表情失败的原因与解决

    这篇文章主要给大家介绍了关于mysql中插入emoji表情失败的原因与解决方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...

    许渺1702019-06-13
  • MysqlMySQL8.0新特性之支持原子DDL语句

    MySQL8.0新特性之支持原子DDL语句

    这MySQL 8.0开始支持原子数据定义语言(DDL)语句。此功能称为原子DDL。这篇文章主要介绍了MySQL8.0新特性——支持原子DDL语句,需要的朋友可以参考下...

    一个笨小孩3462019-07-01
  • Mysql如何修改Mysql中group_concat的长度限制

    如何修改Mysql中group_concat的长度限制

    在mysql中,有个函数叫“group_concat”,平常使用可能发现不了问题,在处理大数据的时候,会发现内容被截取了。怎么解决这一问题呢,下面脚本之家小编给大家带来了Mysql中group_concat的长度限制问题,感兴趣的朋友一起看看吧...

    JouyPub2942019-06-30
  • MysqlMysql中SQL语句不使用索引的情况

    Mysql中SQL语句不使用索引的情况

    今天小编就为大家分享一篇关于Mysql中SQL语句不使用索引的情况,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧...

    CODETC4082019-06-24
北京塞车全天计划精准版