备案控制台
开发者社区 数据库 文章 正文

DataFrame持久存储

2023-05-30 163
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云数据库 RDS MySQL Serverless，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
云数据库 RDS MySQL Serverless，价值2615元额度，1个月
简介： DataFrame持久存储

1. 实验室名称：

大数据实验教学系统

2. 实验项目名称：

DataFrame持久存储

3. 实验学时：

4. 实验原理：

DataFrame数据经过计算以后，可以持久到外部存储中，如关系型数据库和HDFS中。Spark对此提供了支持。

5. 实验目的：

掌握DataFrame存储操作。

6. 实验内容：

将DataFrame持久存储。具体包含如下内容：

  - 写入MySQL

  - 写入HDFS

7. 实验器材（设备、虚拟机名称）：

硬件：x86_64 ubuntu 16.04服务器

  软件：JDK 1.8，Spark-2.3.2，Hadoop-2.7.3，zeppelin-0.8.1

8. 实验步骤：

8.1 环境准备

1、右击Ubuntu操作系统桌面，从弹出菜单中选择【Open in Terminal】命令打开终端。

 在终端窗口下，输入以下命令，分别启动HDFS集群、Spark集群和Zeppelin服务器：


1.  $ start-dfs.sh
2.  $ cd /opt/spark
3.  $ ./sbin/start-all.sh
4.  $ zeppelin-daemon.sh start

2、将本地数据上传至HDFS上。在终端窗口中，分别执行以下命令上传数据：

1.  $ hdfs dfs -mkdir -p /data/dataset/batch
2.  $ hdfs dfs -put /data/dataset/batch/Online-Retail.txt /data/dataset/batch/

3、因为后面的实验中需要访问MySQL数据库，所以先要将MySQL的jdbc驱动程序拷贝到Spark的jars目录下。在终端窗口，执行如下的命令：

1.  $ cp /data/software/mysql-connector-java-5.1.45-bin.jar /opt/spark/jars/

4、启动浏览器，打开zeppelin notebook首页，点击【Create new note】链接，创建一个新的笔记本，名字为【rdd_demo】，解释器默认使用【spark】，如下图所示：


6c94bea2b8894e36b4507ea58ee68394.png

8.2 数据存储

1、读取数据，生成RDD，创建DataFrame。在zeppelin中执行如下代码：

1.  import org.apache.spark.sql.types._
2.  import org.apache.spark.sql._
3.  import org.apache.spark.sql.functions._
4.      　
5.  // 数据路径
6.  var filePath = "/data/dataset/batch/Online-Retail.txt"
7.      　
8.  // 加载RDD
9.  var inFileRDD= sc.textFile(filePath)
10. // 以制表符进行分割
11. var allRowsRDD=inFileRDD.map(x=> x.split("\t"))
12.     　
13. // 获取RDD的第一条数据头标签
14. var header = allRowsRDD.first()
15.     　
16. // 去除标题行
17. var data = allRowsRDD.filter(x => x(0) != header(0))
18.     　
19. // 创建Schema
20. var fields= List(StructField("invoiceNo", StringType, true),
21.                  StructField("stockCode", StringType, true),
22.                  StructField("description", StringType, true),
23.                  StructField("quantity", IntegerType, true),
24.                  StructField("invoiceDate", StringType, true),
25.                  StructField("unitPrice", DoubleType, true),
26.                  StructField("customerID", StringType, true),
27.                  StructField("country", StringType, true)
28.       )
29. val schema = StructType(fields)
30.     　
31. 
32. // 将RDD中的每行数据转换为Row对象
33. var rowRDD = data.map( x => Row(x(0),x(1),x(2),x(3).toInt,x(4),x(5).toDouble,x(6),x(7)))
34.     　
35. // 创建DataFrame
36. var r1DF = spark.createDataFrame(rowRDD,schema)
37. // 显示DataFrame数据
38.  r1DF.show(5)

同时按下【shift+enter】对程序进行输出。输出内容如下所示：

1.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+
2.  |invoiceNo|stockCode|         description|quantity|   invoiceDate|unitPrice|customerID|       country|
3.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+
4.  |   536365|   85123A|WHITE HANGING HEA...|       6|2010/12/1 8:26|     2.55|     17850|United Kingdom|
5.  |   536365|    71053| WHITE METAL LANTERN|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|
6.  |   536365|   84406B|CREAM CUPID HEART...|       8|2010/12/1 8:26|     2.75|     17850|United Kingdom|
7.  |   536365|   84029G|KNITTED UNION FLA...|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|
8.  |   536365|   84029E|RED WOOLLY HOTTIE...|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|
9.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+
10. only showing top 5 rows

2、数据类型转换，创建本地视图，调用sql语句进行查询。在zeppelin中执行如下代码：

1.  // 将invoiceDate列强制转换为时间类型
2.  var ts = unix_timestamp($"invoiceDate","yyyy/MM/dd HH:mm").cast("timestamp")
3.      　
4.  // 为DataFrame添加一列
5.  var r2DF = r1DF.withColumn("ts",ts)
6.  // 显示添加后的数据
7.  r2DF.show(5)

同时按下【shift+enter】对程序进行输出。输出结果如下所示：

1.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+-------------------+
2.  |invoiceNo|stockCode|         description|quantity|   invoiceDate|unitPrice|customerID|       country|                 ts|
3.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+-------------------+
4.  |   536365|   85123A|WHITE HANGING HEA...|       6|2010/12/1 8:26|     2.55|     17850|United Kingdom|2010-12-01 08:26:00|
5.  |   536365|    71053| WHITE METAL LANTERN|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|2010-12-01 08:26:00|
6.  |   536365|   84406B|CREAM CUPID HEART...|       8|2010/12/1 8:26|     2.75|     17850|United Kingdom|2010-12-01 08:26:00|
7.  |   536365|   84029G|KNITTED UNION FLA...|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|2010-12-01 08:26:00|
8.  |   536365|   84029E|RED WOOLLY HOTTIE...|       6|2010/12/1 8:26|     3.39|     17850|United Kingdom|2010-12-01 08:26:00|
9.  +---------+---------+--------------------+--------+--------------+---------+----------+--------------+-------------------+
10. only showing top 5 rows

3、对数据进行查询，选择满足条件的数据。在zeppelin中执行如下代码：

1.  import java.util.Properties
2.      　
3.  // 创建本地临时视图
4.  r2DF.createOrReplaceTempView("retailTable")
5.  // 查找时间小于2011-12-01的数据
6.  var r3DF = spark.sql("select * from retailTable where ts<\"2011-12-01\"")
7.      　
8.  // 查找时间大于等于2011-12-01的数据
9.  var r4DF = spark.sql("select * from retailTable where ts>=\"2011-12-01\"")
10.     　
11. // 选取数据
12. var selectData =  r4DF.select("invoiceNo","stockCode","description","quantity","unitPrice","customerID","country","ts")
13.     　
14. // 修改列的名字
15. var writeMySQL = selectData.withColumnRenamed("ts","invoiceDate")
16. // 显示修改后的DataFrame
17. writeMySQL.show(5)

同时按下【shift+enter】对程序进行输出。输出结果如下所示：

1.  +---------+---------+--------------------+--------+---------+----------+--------------+-------------------+
2.  |invoiceNo|stockCode|         description|quantity|unitPrice|customerID|       country|        invoiceDate|
3.  +---------+---------+--------------------+--------+---------+----------+--------------+-------------------+
4.  |  C579889|    23245|SET OF 3 REGENCY ...|      -8|     4.15|     13853|United Kingdom|2011-12-01 08:12:00|
5.  |  C579890|    84947|ANTIQUE SILVER TE...|      -1|     1.25|     15197|United Kingdom|2011-12-01 08:14:00|
6.  |  C579890|    23374|RED SPOT PAPER GI...|      -1|     0.82|     15197|United Kingdom|2011-12-01 08:14:00|
7.  |  C579890|    84945|MULTI COLOUR SILV...|      -2|     0.85|     15197|United Kingdom|2011-12-01 08:14:00|
8.  |  C579891|    23485|BOTANICAL GARDENS...|      -1|     25.0|     13644|United Kingdom|2011-12-01 08:18:00|
9.  +---------+---------+--------------------+--------+---------+----------+--------------+-------------------+
10. only showing top 5 rows

4、将查询到的数据存储到MySQL中。

 (1)首先启动MySQL服务器。在终端窗口中，执行以下命令：

1.  $ service mysql start

(2)登录MySQL服务器。在终端窗口中，执行以下命令：

1.  $ mysql -u root -p

然后根据提示，输入登录密码：root。

 (3)执行以下SQL语句，创建测试表：

1.  mysql> create database retailDB;
2.  mysql> exit;

(4)在zeppelin中执行如下代码：


1.  // 将DataFrame数据存储到数据库中
2.  val prop = new Properties()
3.  prop.setProperty("user", "root")
4.  prop.setProperty("password", "root")
5.      　
6.  writeMySQL.write.mode("append").jdbc("jdbc:mysql://localhost:3306/retailDB?characterEncoding=UTF-8", "transactions", prop)

验证保存成功。进入MySQL，通过【select count(*) from transactions;】来查看写入MySQL的数据条数。


5、将DataFrame存储到HDFS中。选择满足条件的数据，将数据写入到HDFS中。在zeppelin中执行如下代码：

1.  var selectData = r3DF.select("invoiceNo","stockCode","description","quantity","unitPrice","customerID","country","ts")
2.  var writeHDFS = selectData.withColumnRenamed("ts","invoiceDate")
3.  writeHDFS.select("*").write.format("json").save("/Users/r3DF")

同时按下【shift+enter】对程序进行输出。

 验证保存到HDFS中成功。在终端窗口下，执行以下命令，查看写入HDFS的json数据：

1.  # hdfs dfs -ls /Users/r3DF

可以看到已经写入成功，如下图所示：


1013f4ddbc2040558b6b69c9ff11f890.png


9. 实验结果及分析：

实验结果运行准确，无误


10. 实验结论：

经过本节实验的学习，通过学习DataFrame持久存储，进一步巩固了我们的Spark基础。


11. 总结及心得体会：

SparkSQL对SQL语句的处理和关系型数据库采用了类似的方法，SparkSQL会先将SQL语句进行解析Parse形成一个Tree，然后使用Rule对Tree进行绑定、优化等处理过程，通过模式匹配对不同类型的节点采用不同的操作。而SparkSQL的查询优化器是Catalyst，它负责处理查询语句的解析、绑定、优化和生成物理计划等过程，Catalyst是SparkSQL最核心的部分，其性能优劣将决定整体的性能。


12、 实验测试

1、数据写入MySQL中mode=’append’的意思是什么( A ){单选}

 A、追加

 B、覆盖

 C、修改

 D、删除


13、实验拓展

1、给定给一个文本数据，将数据转换为DataFrame类型，并将数据写入到MySQL 中。


34f2257444904fb4be7d73efd02e4438.png

文章标签：
云数据库 RDS MySQL 版
Java
分布式计算
SQL
关系型数据库
Ubuntu
数据安全/隐私保护
MySQL
Spark
存储
数据库连接
相关实践学习
基于CentOS快速搭建LAMP环境
本教程介绍如何搭建LAMP环境，其中LAMP分别代表Linux、Apache、MySQL和PHP。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心，利用清洁计算为万物互联的新世界提供源源不断的能源动力，目前开服的区域包括中国（华北、华东、华南、香港）、新加坡、美国（美东、美西）、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程，来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品：云服务器ECS 云服务器 ECS（Elastic Compute Service）是一种弹性可伸缩的计算服务，助您降低 IT 成本，提升运维效率，使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
不懂开发的程序猿
目录
相关文章
倔强的大蜗牛
|
25天前
|
存储 小程序
数据的存储
数据的存储
倔强的大蜗牛
32 1
ForcibleBugMaker
|
1月前
|
存储 C语言
数据在内存中的存储2
数据在内存中的存储2
ForcibleBugMaker
30 1
YoungMLet
|
7月前
|
存储
【数据的存储】
【数据的存储】
YoungMLet
47 0
醋溜马桶圈
|
6月前
|
存储 编译器 C语言
Day_11 ＞ 数据在内存中的存储
Day_11 ＞ 数据在内存中的存储
醋溜马桶圈
35 0
EDG.TSWS
|
6月前
|
存储 编译器 C++
剖析数据在内存中的存储（下）
剖析数据在内存中的存储（下）
EDG.TSWS
18 0
EDG.TSWS
|
6月前
|
存储 编译器 C语言
剖析数据在内存中的存储（上）
剖析数据在内存中的存储（上）
EDG.TSWS
22 0
Dream_Chaser～
|
7月前
|
存储
数据的存储（下）
数据的存储（下）
Dream_Chaser～
31 0
Dream_Chaser～
|
7月前
|
存储 小程序 编译器
数据的存储（上）
数据的存储（上）
Dream_Chaser～
26 0
Dio夹心小面包
|
7月前
|
存储 C语言
数据在内存中的存储（剖析）
数据在内存中的存储（剖析）
Dio夹心小面包
36 0
邹邹很busy。
|
9月前
|
存储 应用服务中间件 nginx
k8s--数据存储、EmptyDir存储
k8s--数据存储、EmptyDir存储
邹邹很busy。
160 0

热门文章

最新文章

  • 1
    企业应用架构实践（复杂性应对之道）
  • 2
    SolidWorks2021中文版【Solid Works 附补丁】完美破解版下载 安装教程
  • 3
    3+1保障：高可用系统稳定性是如何炼成的？
  • 4
    大数据上手实战！《Elasticsearch 实战进阶营》第二季限时免费报名啦
  • 5
    OLAP与OLTP介绍
  • 6
    浅显易懂的机器学习（八）—— 随机森林分类
  • 7
    Windows 2008 AD 备份
  • 8
    《Ext JS权威指南》——2.9节为本书示例准备一个模板
  • 9
    NS2原理
  • 10
    Asp.net MVC2.0系列文章-显示列表和详细页面操作
  • 1
    Java中数组详解
    40
  • 2
    智能评估时代：SurveyKing开源问卷系统YYDS
    62
  • 3
    LabVIEW编程LabVIEW开发高级数据采集技术 操作数字IO 例程与相关资料
    37
  • 4
    【大模型】描述与 LLM 相关的个人项目或感兴趣的领域
    37
  • 5
    【大模型】使用哪些资源来了解 LLM 的最新进展？
    47
  • 6
    LabVIEW编程LabVIEW开发高级数据采集技术 计数器定时器的操作 例程与相关资料
    24
  • 7
    LabVIEW编程LabVIEW开发高级数据采集技术 模拟波形的生成 例程与相关资料
    22
  • 8
    Linux：进程等待 & 进程替换
    27
  • 9
    LabVIEW编程LabVIEW开发高级数据采集技术 同步 例程与相关资料
    19
  • 10
    LabVIEW编程LabVIEW开发高级数据采集技术定时与触发 例程与相关资料
    17

    • 相关电子书

    更多
  • HBase2.0重新定义小对象实时存取
  • 低代码开发师（初级）实战教程
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）