【Spark分布式内存计算框架——Spark Core】7. RDD Checkpoint、外部数据源

news/2024/5/20 1:12:56 标签: spark, 分布式, 大数据, hdfs, hadoop

第五章 RDD Checkpoint

RDD 数据可以持久化，但是持久化/缓存可以把数据放在内存中，虽然是快速的，但是也是最不可靠的；也可以把数据放在磁盘上，也不是完全可靠的！例如磁盘会损坏等。

Checkpoint的产生就是为了更加可靠的数据持久化，在Checkpoint的时候一般把数据放在HDFS上，这就天然的借助了HDFS天生的高容错、高可靠来实现数据最大程度上的安全，实现了RDD的容错和高可用。

在Spark Core中对RDD做checkpoint，可以切断做checkpoint RDD的依赖关系，将RDD数据保存到可靠存储（如HDFS）以便数据恢复；

在这里插入图片描述
演示范例代码如下：

import org.apache.spark.{SparkConf, SparkContext}
/**
* RDD数据Checkpoint设置，案例演示
*/
object SparkCkptTest {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// TODO: 设置检查点目录，将RDD数据保存到那个目录
sc.setCheckpointDir("datas/spark/ckpt/")
// 读取文件数据
val datasRDD = sc.textFile("datas/wordcount/wordcount.data")
// TODO: 调用checkpoint函数，将RDD进行备份，需要RDD中Action函数触发
datasRDD.checkpoint()
datasRDD.count()
// TODO: 再次执行count函数, 此时从checkpoint读取数据
datasRDD.count()
// 应用程序运行结束，关闭资源
Thread.sleep(100000)
sc.stop()
}
}

持久化和Checkpoint的区别：
1）、存储位置

  • Persist 和 Cache 只能保存在本地的磁盘和内存中(或者堆外内存)；
  • Checkpoint 可以保存数据到 HDFS 这类可靠的存储上；
    2）、生命周期
  • Cache和Persist的RDD会在程序结束后会被清除或者手动调用unpersist方法；
  • Checkpoint的RDD在程序结束后依然存在，不会被删除；
    3）、Lineage(血统、依赖链、依赖关系)
  • Persist和Cache，不会丢掉RDD间的依赖链/依赖关系，因为这种缓存是不可靠的，如果出现了一些错误(例如 Executor 宕机)，需要通过回溯依赖链重新计算出来；
  • Checkpoint会斩断依赖链，因为Checkpoint会把结果保存在HDFS这类存储中，更加的安全可靠，一般不需要回溯依赖链；
    在这里插入图片描述

第六章 外部数据源

Spark可以从外部存储系统读取数据，比如RDBMs表中或者HBase表中读写数据，这也是企业中常常使用，如下两个场景：
1）、要分析的数据存储在HBase表中，需要从其中读取数据数据分析

  • 日志数据：电商网站的商家操作日志
  • 订单数据：保险行业订单数据
    2）、使用Spark进行离线分析以后，往往将报表结果保存到MySQL表中
  • 网站基本分析（pv、uv。。。。。）

6.1 HBase 数据源

Spark可以从HBase表中读写（Read/Write）数据，底层采用TableInputFormat和TableOutputFormat方式，与MapReduce与HBase集成完全一样，使用输入格式InputFormat和输出格式OutputFoamt。
在这里插入图片描述

HBase Sink
回 顾 MapReduce 向 HBase 表 中 写 入 数 据 ， 使 用 TableReducer ， 其 中 OutputFormat 为TableOutputFormat，读取数据Key：ImmutableBytesWritable，Value：Put。

写 入 数 据 时 ， 需 要 将 RDD 转换为 RDD[(ImmutableBytesWritable, Put)] 类 型 ， 调 用saveAsNewAPIHadoopFile方法数据保存至HBase表中。

HBase Client连接时，需要设置依赖Zookeeper地址相关信息及表的名称，通过Configuration设置属性值进行传递。
在这里插入图片描述
范例演示：将词频统计结果保存HBase表，表的设计
在这里插入图片描述
代码如下：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* 将RDD数据保存至HBase表中
*/
object SparkWriteHBase {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// TODO: 1、构建RDD
val list = List(("hadoop", 234), ("spark", 3454), ("hive", 343434), ("ml", 8765))
val outputRDD: RDD[(String, Int)] = sc.parallelize(list, numSlices = 2)
// TODO: 2、将数据写入到HBase表中, 使用saveAsNewAPIHadoopFile函数，要求RDD是(key, Value)
// TODO: 组装RDD[(ImmutableBytesWritable, Put)]
/**
* HBase表的设计：
* 表的名称：htb_wordcount
* Rowkey: word
* 列簇: info
* 字段名称： count
*/
val putsRDD: RDD[(ImmutableBytesWritable, Put)] = outputRDD.mapPartitions{ iter =>
iter.map { case (word, count) =>
// 创建Put实例对象
val put = new Put(Bytes.toBytes(word))
// 添加列
put.addColumn(
// 实际项目中使用HBase时，插入数据，先将所有字段的值转为String，再使用Bytes转换为字节数组
Bytes.toBytes("info"), Bytes.toBytes("cout"), Bytes.toBytes(count.toString)
)
// 返回二元组
(new ImmutableBytesWritable(put.getRow), put)
}
}
// 构建HBase Client配置信息
val conf: Configuration = HBaseConfiguration.create()
// 设置连接Zookeeper属性
conf.set("hbase.zookeeper.quorum", "node1.itcast.cn")
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("zookeeper.znode.parent", "/hbase")
// 设置将数据保存的HBase表的名称
conf.set(TableOutputFormat.OUTPUT_TABLE, "htb_wordcount")
/*
def saveAsNewAPIHadoopFile(
path: String,// 保存的路径
keyClass: Class[_], // Key类型
valueClass: Class[_], // Value类型
outputFormatClass: Class[_ <: NewOutputFormat[_, _]], // 输出格式OutputFormat实现
conf: Configuration = self.context.hadoopConfiguration // 配置信息
): Unit
*/
putsRDD.saveAsNewAPIHadoopFile(
"datas/spark/htb-output-" + System.nanoTime(), //
classOf[ImmutableBytesWritable], //
classOf[Put], //
classOf[TableOutputFormat[ImmutableBytesWritable]], //
conf
)
// 应用程序运行结束，关闭资源
sc.stop()
}
}

运行完成以后，使用hbase shell查看数据：
在这里插入图片描述
HBase Source
回 顾 MapReduce 从 读 HBase 表 中 的 数 据 ， 使 用 TableMapper ， 其 中 InputFormat 为TableInputFormat，读取数据Key：ImmutableBytesWritable，Value：Result。

从HBase表读取数据时，同样需要设置依赖Zookeeper地址信息和表的名称，使用Configuration设置属性，形式如下：
在这里插入图片描述

此外，读取的数据封装到RDD中，Key和Value类型分别为：ImmutableBytesWritable和Result,不支持Java Serializable导致处理数据时报序列化异常。设置Spark Application使用Kryo序列化，性能要比Java 序列化要好，创建SparkConf对象设置相关属性，如下所示：
在这里插入图片描述
范例演示：从HBase表读取词频统计结果，代码如下

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration}
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* 从HBase 表中读取数据，封装到RDD数据集
*/
object SparkReadHBase {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// TODO: 设置使用Kryo 序列化方式
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// TODO: 注册序列化的数据类型
.registerKryoClasses(Array(classOf[ImmutableBytesWritable], classOf[Result]))
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// TODO: a. 读取HBase Client 配置信息
val conf: Configuration = HBaseConfiguration.create()
conf.set("hbase.zookeeper.quorum", "node1.itcast.cn")
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("zookeeper.znode.parent", "/hbase")
// TODO: b. 设置读取的表的名称
conf.set(TableInputFormat.INPUT_TABLE, "htb_wordcount")
/*
def newAPIHadoopRDD[K, V, F <: NewInputFormat[K, V]](
conf: Configuration = hadoopConfiguration,
fClass: Class[F],
kClass: Class[K],
vClass: Class[V]
): RDD[(K, V)]
*/
val resultRDD: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(
conf, //
classOf[TableInputFormat], //
classOf[ImmutableBytesWritable], //
classOf[Result] //
)
println(s"Count = ${resultRDD.count()}")
resultRDD
.take(5)
.foreach { case (rowKey, result) =>
println(s"RowKey = ${Bytes.toString(rowKey.get())}")
// HBase表中的每条数据封装在result对象中，解析获取每列的值
result.rawCells().foreach { cell =>
val cf = Bytes.toString(CellUtil.cloneFamily(cell))
val column = Bytes.toString(CellUtil.cloneQualifier(cell))
val value = Bytes.toString(CellUtil.cloneValue(cell))
val version = cell.getTimestamp
println(s"\t $cf:$column = $value, version = $version")
}
}
// 应用程序运行结束，关闭资源
sc.stop()
}
}

运行结果：
在这里插入图片描述

6.2 MySQL 数据源

实际开发中常常将分析结果RDD保存至MySQL表中，使用foreachPartition函数；此外Spark中提供JdbcRDD用于从MySQL表中读取数据。

调用RDD#foreachPartition函数将每个分区数据保存至MySQL表中，保存时考虑降低RDD分区数目和批量插入，提升程序性能。

范例演示：将词频统计WordCount结果保存MySQL表tb_wordcount。

  • 建表语句
USE db_test ;
CREATE TABLE `tb_wordcount` (
`count` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL,
`word` varchar(100) NOT NULL,
PRIMARY KEY (`word`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci ;
  • 演示代码
import java.sql.{Connection, DriverManager, PreparedStatement}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* 将词频统计结果保存到MySQL表中
*/
object SparkWriteMySQL {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 1. 从HDFS读取文本数据，封装集合RDD
val inputRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data")
// 2. 处理数据，调用RDD中函数
val resultRDD: RDD[(String, Int)] = inputRDD
// 3.a 每行数据分割为单词
.flatMap(line => line.split("\\s+"))
// 3.b 转换为二元组，表示每个单词出现一次
.map(word => (word, 1))
// 3.c 按照Key分组聚合
.reduceByKey((tmp, item) => tmp + item)
// 3. 输出结果RDD保存到MySQL数据库
resultRDD
// 对结果RDD保存到外部存储系统时，考虑降低RDD分区数目
.coalesce(1)
// 对分区数据操作
.foreachPartition{iter => saveToMySQL(iter)}
// 应用程序运行结束，关闭资源
sc.stop()
}
/**
* 将每个分区中的数据保存到MySQL表中
* @param datas 迭代器，封装RDD中每个分区的数据
*/
def saveToMySQL(datas: Iterator[(String, Int)]): Unit = {
// a. 加载驱动类
Class.forName("com.mysql.cj.jdbc.Driver")
// 声明变量
var conn: Connection = null
var pstmt: PreparedStatement = null
try{
// b. 获取连接
conn = DriverManager.getConnection(
"jdbc:mysql://node1.itcast.cn:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnic
ode=true",
"root", "123456"
)
// c. 获取PreparedStatement对象
val insertSql = "INSERT INTO db_test.tb_wordcount (word, count) VALUES(?, ?)"
pstmt = conn.prepareStatement(insertSql)
conn.setAutoCommit(false)
// d. 将分区中数据插入到表中，批量插入
datas.foreach{case (word, count) =>
pstmt.setString(1, word)
pstmt.setLong(2, count.toLong)
// 加入批次
pstmt.addBatch()
}
// TODO: 批量插入
pstmt.executeBatch()
conn.commit()
}catch {
case e: Exception => e.printStackTrace()
}finally {
if(null != pstmt) pstmt.close()
if(null != conn) conn.close()
}
}
}
  • 运行程序，查看数据库表的数据
    在这里插入图片描述

http://www.niftyadmin.cn/n/56829.html

相关文章

ROS小车研究笔记2/11/2023：使用ssh远程登录小车

ROS小车研究笔记2/11/2023：使用ssh远程登录小车

1 SSH简介&#xff1a; SSH全称Secure Shell&#xff0c;是一种建立在应用层的安全网络协议。其安全性又非对称加密(RSA)实现 对称加密&#xff1a;使用同一密钥对信息进行加密和解密&#xff0c;但是一旦该密钥被窃取就会威胁通信安全 非对称加密&#xff1a;使用公钥和私钥。…
阅读更多...
[标准库]STM32F103R8T6 高级定时器--PWM输出和带死区互补PWM输出

[标准库]STM32F103R8T6 高级定时器--PWM输出和带死区互补PWM输出

前言 STM32F103系列的MCU&#xff0c;相比普通的51单片机&#xff0c;在输出硬件PWM这个功能上要强不少&#xff0c;两者实现的方式都类似&#xff0c;都是通过一个定时器来启用硬件PWM输出&#xff0c;不过在输出PWM通道的数量上&#xff0c;32F103要强上不少。仅通过一个高级…
阅读更多...
剑指offer（中等）

剑指offer（中等）

目录 二维数组中的查找 重建二叉树 矩阵中的路径 剪绳子 剪绳子② 数值的整数次方 表示数值的字符串 树的子结构 栈的压入、弹出序列 从上到下打印二叉树① 从上到下打印二叉树③ 二叉搜索树的后序遍历序列 二叉树中和为某一值的路径 复杂链表的复制 二叉搜索树与…
阅读更多...
Android Jetpack组件之WorkManager高级概念介绍与使用(三)

Android Jetpack组件之WorkManager高级概念介绍与使用(三)

一、介绍 通过前面两篇&#xff0c;我们基本掌握了组件的workmanager的接入&#xff0c;以及api的使用等。但是一个框架如果运用在复杂的项目中&#xff0c;肯定需要有其他额外的支持&#xff0c;介绍来我们将会介绍高级概念&#xff0c;以及对前面的知识点进行回顾与拓展。 高…
阅读更多...
自定义软件帮助文档（qt assistant实现）

自定义软件帮助文档（qt assistant实现）

网上搜了一下&#xff0c;软件的帮助文档&#xff0c;三个都可以&#xff1a;https://github.com/zealdocs/zeal&#xff0c;https://zealdocs.org/&#xff0c;看看这个博客说的 https://blog.csdn.net/libaineu2004/article/details/125028913&#xff0c;这个也是开源的&…
阅读更多...
window11 安装node及配置环境变量

window11 安装node及配置环境变量

一、安装环境 本教程演示的环境&#xff1a; 系统&#xff1a;win 11 64位 node.js下载地址: http://nodejs.cn/ node.js版本&#xff1a;长期支持版本&#xff08;本教程基于16.15.0&#xff09; 点击选中图标下载到电脑本地即可。 二、安装步骤 1、双击安装包&#xff0c;一…
阅读更多...
【吉先生的Java全栈之路】

【吉先生的Java全栈之路】

吉士先生Java全栈学习路线&#x1f9e1;第一阶段Java基础: 在第一阶段:我们要认真听讲,因为基础很重要!基础很重要!基础很重要!!! 重要的事情说三遍。在这里我们先学JavaSE路线&#xff1b;学完之后我们要去学第一个可视化组件编程《GUI》&#xff1b;然后写个《贪吃蛇》游戏耍…
阅读更多...
ChatGPT的注册与使用方法

ChatGPT的注册与使用方法

ChatGPT无疑是最近最火的一个大模型AI应用，当艺术家还在跟AI争论AI绘画的问题时，NLP领域中的ChatGPT大模型应用，仿佛一下子替代了AI绘画应用，成为了最新争论的话题。最近也看到很多人在讨论ChatGPT，当然，很多人就算不是人工智能领域的也开始讨论ChatGPT，可见ChatGPT的风…
阅读更多...
最新文章

Copyright @ 2022~2023