• SparkCore系列-8、RDD 读写外部数据源

    版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

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

    官方网址：http://spark.apache.org/、https://databricks.com/spark/about
    在这里插入图片描述

    目录

    回顾

    上篇文章介绍了一个小案例，对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()
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76

    运行完成以后，使用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()
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64

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

    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 ;

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    演示代码

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77

    运行程序，查看数据库表的数据
    在这里插入图片描述

    下回分解

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

  • 相关阅读:
    基于springboot实现摄影跟拍预定管理系统【项目源码+论文说明】
    分布式事务解决方案Seata-Golang浅析
    640. 求解方程(JavaScript)
    采用connector-c++ 8.0操作数据库
    Spring Cloud Alibaba微服务从入门到进阶（四）（服务发现-Nacos ）
    状态压缩DP
    05 vue 计算属性的练习 1 methods \computed\ watch三种方法对比实例
    Anemoi hash:一种SNARK-friendly的哈希函数
    SpringMVC之CRUD------增删改查
    主动人机交互与被动人机交互
  • 原文地址：https://blog.csdn.net/l848168/article/details/126054165