前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >SparkSQL如何实现多数据源交互?这篇博客或许能告诉你答案!

SparkSQL如何实现多数据源交互?这篇博客或许能告诉你答案!

作者头像
大数据梦想家
发布2021-01-27 16:02:37
6240
发布2021-01-27 16:02:37
举报

学了一段时间的SparkSQL,相信大家都已经知道了SparkSQL是一个相当强大的存在,它在一个项目的架构中扮演着离线数据处理的"角色",相较于前面学过的HQL,SparkSQL能明显提高数据的处理效率。正因为如此,SparkSQL就会涉及到与多种的数据源进行一个交互的过程。那到底是如何交互的呢,下文或许能给你带来答案…

码字不易,先赞后看,养成习惯!

在这里插入图片描述
在这里插入图片描述

Spark SQL可以与多种数据源进行交互,如普通文本jsonparquetcsvMySQL

下面将从写数据和读数据两个角度来进行演示。

准备数据

以下面的演示为例,我们在本地的D:\data目录下创建一个person.txt

代码语言:javascript
复制
19 zhhshang 66
20 lisi 66
19 wangwu 77
31 zhaoliu 66
19 maqi 88

并在本地Mysql创建一个数据库spark_test,并创建一个表名persons,并且表结构如下所示:

在这里插入图片描述
在这里插入图片描述

写入数据

代码语言:javascript
复制
object WriterDataSourceDemo {
  case class Person(id:Int,name:String,age:Int)

  def main(args: Array[String]): Unit = {
    //1.创建SparkSession
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2.读取文件
    val fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
    val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
    val rowRDD: RDD[Person] = linesRDD.map(line =>Person(line(0).toInt,line(1),line(2).toInt))
    //3.将RDD转成DF
    //注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
    import spark.implicits._
    //注意:上面的rowRDD的泛型是Person,里面包含了Schema信息
    //所以SparkSQL可以通过反射自动获取到并添加给DF
    val personDF: DataFrame = rowRDD.toDF
    //==================将DF写入到不同数据源===================
    //Text data source supports only a single column, and you have 3 columns.;
    //personDF.write.text("D:\\data\\output\\text")
    personDF.write.json("D:\\data\\output\\json")
    personDF.write.csv("D:\\data\\output\\csv")
    personDF.write.parquet("D:\\data\\output\\parquet")
    val prop = new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","root")
    // 将数据写入到数据库
    personDF.write.mode(SaveMode.Overwrite).jdbc(
      "jdbc:mysql://localhost:3306/spark_test?characterEncoding=UTF-8","persons",prop)
    println("写入成功")
    sc.stop()
    spark.stop()
  }
}

运行结果:

我们在程序中设置的输出路径下看到了已经生成的三个文件

在这里插入图片描述
在这里插入图片描述

csv目录

在这里插入图片描述
在这里插入图片描述

json目录

在这里插入图片描述
在这里插入图片描述

parquet目录

在这里插入图片描述
在这里插入图片描述

再让我们打开数据库看看

在这里插入图片描述
在这里插入图片描述

发现我们新建的数据库中的数据也添加了进来

说明我们的数据写入成功了,感兴趣的朋友们可以自己试一下哟~

下面我们再来尝试把数据从我们写入的数据文件中读取出来。

读数据

代码语言:javascript
复制
object ReadDataSourceDemo {
  def main(args: Array[String]): Unit = {
    //1.创建SparkSession
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2.读取文件
    spark.read.json("D:\\data\\output\\json").show()
    spark.read.csv("D:\\data\\output\\csv").toDF("id","name","age").show()
    spark.read.parquet("D:\\data\\output\\parquet").show()
    val prop = new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","root")
    spark.read.jdbc(
      "jdbc:mysql://localhost:3306/spark_test?characterEncoding=UTF-8","persons",prop).show()
    sc.stop()
    spark.stop()
  }
}

运行结果

在这里插入图片描述
在这里插入图片描述

看到上图的结果说明我们成功实现了将数据导出,再读取的过程。

总结

  1. SparkSQL 写数据: DataFrame/DataSet.write.json/csv/jdbc
  2. SparkSQL读数据 SparkSession.read.json/csv/text/jdbc/format

结语

本次的分享就到这里,受益的朋友或对大数据技术感兴趣的伙伴可以点个赞关注一下博主,后续会持续更新大数据的相关内容,敬请期待(✪ω✪)

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2020-04-12 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 准备数据
  • 写入数据
  • 读数据
  • 总结
  • 结语
相关产品与服务
文件存储
文件存储(Cloud File Storage,CFS)为您提供安全可靠、可扩展的共享文件存储服务。文件存储可与腾讯云服务器、容器服务、批量计算等服务搭配使用,为多个计算节点提供容量和性能可弹性扩展的高性能共享存储。腾讯云文件存储的管理界面简单、易使用,可实现对现有应用的无缝集成;按实际用量付费,为您节约成本,简化 IT 运维工作。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档