大数据 hbase hadoop SparkCore系列-8、RDD 读写外部数据源

传送门：大数据系列文章目录

官方网址：http://spark.apache.org/、https://databricks.com/spark/about

回顾场景介绍HBase 数据源HBase SinkHBase Source

MySQL 数据源下回分解

回顾

上篇文章介绍了一个小案例，对sogou的日志进行分析，虽然东西不多，但是也五脏俱全了。

场景

之前的代码我们都是从本地读取数据，跟企业中真实的开发还是不太一样，企业中都是从外部读取数据源，之后把分析结果写入到Mysql或者其他的外部数据源，现在就来真实的操作下是如何从外部数据源读取数据或者把数据写入到外部数据源。

介绍

Spark可以从外部存储系统读取数据，比如RDBMs表中或者HBase表中读写数据，这也是企业中常常使用，如下两个场景：

1）要分析的数据存储在HBase表中，需要从其中读取数据数据分析

日志数据：电商网站的商家操作日志订单数据：保险行业订单数据

2）使用Spark进行离线分析以后，往往将报表结果保存到MySQL表中

网站基本分析（pv、 uv。。。。。）

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)

}

// 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")

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.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)

}

// TODO: a. 读取HBase Client 配置信息

val conf: Configuration = HBaseConfiguration.create()

conf.set("hbase.zookeeper.quorum", "node1")

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 (_, result) =>

println(s"RowKey = ${Bytes.toString(result.getRow)}")

// 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()

}

运行结果：

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)

}

// 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:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=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()

}

运行程序，查看数据库表的数据

下回分解

外部数据源结束后，下篇文章就是介绍共享变量了操作了。

参考阅读

评论可见，请评论后查看内容，谢谢！！！

您阅读本篇文章共花了：

金钥匙

大数据 hbase hadoop SparkCore系列-8、RDD 读写外部数据源

Linux多虚拟机集群化配置详解(Zookeeper集群、Kafka集群、Hadoop集群、HBase集群、Spark集群、Flink集群、Zabbix、Grafana部署)

Hadoop 高可用集群完全分布式安装教程一篇就够用（zookeeper、spark、hbase、mysql、hive)

发表评论取消回复

金钥匙

大数据 hbase hadoop SparkCore系列-8、RDD 读写外部数据源

Linux多虚拟机集群化配置详解(Zookeeper集群、Kafka集群、Hadoop集群、HBase集群、Spark集群、Flink集群、Zabbix、Grafana部署)

Hadoop 高可用集群完全分布式安装教程一篇就够用（zookeeper、spark、hbase、mysql、hive)

相关文章

发表评论取消回复