Learning Spark——Spark连接MysqlmapPartitions高效连接HBase

Posted Trigl

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Learning Spark——Spark连接MysqlmapPartitions高效连接HBase相关的知识,希望对你有一定的参考价值。

执行Spark任务免不了从多个数据源拿数据,除了从HDFS获取数据以外,我们还经常从mysql和HBase中拿数据,今天讲一下如何使用Spark查询Mysql和HBase

1. Spark查询Mysql

首先,Spark连接Mysql当然需要有Mysql的驱动包,你可以在启动时加上如下命令:

bin/spark-shell --driver-class-path /home/hadoop/jars/mysql-connector-java-5.1.34.jar --jars /home/hadoop/jars/mysql-connector-java-5.1.34.jar

还有一种更方便的方法就是直接将这个jar包放到Spark放jar包的目录下面,我的目录是/data/install/spark-2.0.0-bin-hadoop2.7/jars,这样Spark就可以直接找到Mysql驱动包了

然后给出Spark读Mysql时的标准代码:

    val imeis = spark.read.format("jdbc").options(
      Map("url" -> DbUtil.IMEI_DB_URL,
//        "dbtable" -> "(SELECT id,imei,imeiid FROM t_imei_all) a",
        "dbtable" -> DbUtil.IMEI_ALL_TABLE,
        "user" -> DbUtil.IMEI_DB_USERNAME,
        "password" -> DbUtil.IMEI_DB_PASSWORD,
        "driver" -> "com.mysql.jdbc.Driver",
//        "fetchSize" -> "1000",
        "partitionColumn" -> "id",
        "lowerBound" -> "1",
        "upperBound" -> "15509195",
        "numPartitions" -> "20")).load()

解释一下这段代码:

1、其中spark就是SparkSession

2、如果是读操作,就是spark.read,如果是写操作,就是spark.write

3、options里面就是我们需要查询的一些具体信息,有关配置如下:

配置项含义
url数据库连接地址,如:jdbc:mysql://localhost:3306/IMEI?useUnicode=true&characterEncoding=UTF8
dbtable要查询的表,这里有两种书写方式,一种可以直接写一个表名,如:t_test;另一种是写一条查询语句,但是注意要给一个别名,如:(SELECT id,imei,imeiid FROM t_imei_all) a。建议使用第一种,第二种查询会很慢
driver数据库驱动,如Mysql的是:com.mysql.jdbc.Driver
user数据库用户名
password数据库密码
partitionColumn, lowerBound, upperBound, numPartitions这几个参数用来指定用哪个列来分区,当我们查询的量很大时,例如超过千万的数据量,如果Spark不分区查询的话很快就会报OOM异常了。而且这几个参数只要指定其中一个,其他的就也要指定,partitionColumn是要分区的列,必须是整数类型;lowerBound和upperBound是分区的上下限;numPartitions是分区数
fetchsize用于读操作,每次读取多少条记录
batchsize用于写操作,每次写入多少条记录
isolationLevel用于写操作,数据库的隔离级别
truncate用于写操作,当Spark要执行覆盖表操作时,即启用了SaveMode.Overwrite,使用truncate比使用drop或者recreate操作更高效,默认是false
createTableOptions用于写操作,可以指定建表的语句,如: CREATE TABLE t (name string) ENGINE=InnoDB

4、最后的结果是一个DataFrame,可以很方便地使用SparkSql继续其他操作

2. 如何在RDD中高效连接HBase

连接HBase直接用HBase的API就好了,我们这里重点讲的是在RDD中连接HBase,大家都知道Spark处理的都是很大的数据量,而RDD连接HBase的时候势必会产生很多与HBase的连接,这样很快就会用光连接数,这里我们使用一个算子mapPartitions来解决这个问题

mapPartitions函数和map函数类似,只不过映射函数的参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器,返回的结果也是每一个分区的迭代器。如果在映射的过程中需要频繁创建额外的对象,使用mapPartitions要比map高效的多

比如,将RDD中的所有数据通过JDBC连接写入数据库,如果使用map函数,可能要为每一个元素都创建一个connection,这样开销很大,如果使用mapPartitions,那么只需要针对每一个分区建立一个connection

下面上栗子:RDD中的内容是字符串,这段字符串是HBase主键rowKey的一部分,现在要根据这条字符串查出HBase中的一条信息。例如HBase的主键是aaabbb111222,RDD中存的内容是aaa,我们要查出HBase主键前缀是aaa的所有记录的主键、包名列表和时间

代码如下:

package com.trigl.spark.main

import com.trigl.spark.util.HbaseUtil
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.Result, Scan
import org.apache.hadoop.hbase.util.Bytes
import org.apache.log4j.Level, Logger
import org.apache.spark.SparkConf, SparkContext

/**
  * 高效连接HBase示例
  * created by Trigl at 2017-05-20 15:34
  */
object HBaseDemo 

  def main(args: Array[String]) 
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sparkConf = new SparkConf().setAppName("HBaseDemo")
    val sc = new SparkContext(sparkConf)

    val data = sc.textFile("/test/imei.txt").mapPartitions(getHBaseInfo)

    // 结果以 主键|包列表|时间 的格式存入HDFS
    data.map(l => 
      val rowKey = l._2._1
      val pkgList = l._2._2
      val time = l._1
      rowKey + "|" + pkgList + "|" + time // 用"|"分隔
    ).repartition(1).saveAsTextFile("/test/fenxi/cpz")

    sc.stop()
  

  /**
    * 从HBase查询
    *
    * @param iter mapPartion算子的参数是Iterator
    * @return 返回的也是Iterator
    */
  def getHBaseInfo(iter: Iterator[String]): Iterator[(String, (String, String))] = 

    var pkgList = List[(String, (String, String))]() // 结果格式为(日期,(主键,包名集合))

    // 建立连接查询表
    val conn = HbaseUtil.getConnection(HbaseUtil.TABLE_NAME_CPZ_APP)
    val table = conn.getTable(TableName.valueOf(HbaseUtil.TABLE_NAME_CPZ_APP))

    // 新建Scan用于指定查询内容
    val scan = new Scan()
    scan.setCaching(10000)
    scan.setCacheBlocks(false)
    // 要查询的列
    scan.addColumn(HbaseUtil.COLUMN_FAMILY.getBytes, "packagelist".getBytes)
    scan.addColumn(HbaseUtil.COLUMN_FAMILY.getBytes, "cdate".getBytes)

    while (iter.hasNext) 
      // 要查询的前缀
      val imei = iter.next()
      // HBase前缀查询
      scan.setRowPrefixFilter(imei.getBytes)
      // 查询结果
      val resultScanner = table.getScanner(scan)
      val it = resultScanner.iterator()
      if (it.hasNext) 
        val result: Result = it.next()

        // 主键
        val key = Bytes.toString(result.getRow)
        // 日期
        val cdate = Bytes.toString(result.getValue(HbaseUtil.COLUMN_FAMILY.getBytes, "cdate".getBytes))
        // 包列表
        val packagelist = Bytes.toString(result.getValue(HbaseUtil.COLUMN_FAMILY.getBytes, "packagelist".getBytes))

        // 添加到集合中
        pkgList.::=(cdate, (key, packagelist))
      
    

    // 关闭HBase连接
    table.close()
    conn.close()

    // 结果返回iterator
    pkgList.iterator
  

最后存的结果是:

8643960350683910864396035068383020170421134920XHt3IGTdqIKtV5Y|2627,com.sogou.activity.src,fc8dbdce14c111859fd0111b03e80cd7,0;2856,com.qihoo.appstore,aa90bca1ab548eadd44a0c1d8c34cbda,0|2017-04-21 13:49:18

上面的完整代码见我的github:

https://github.com/Trigl/SparkLearning/blob/master/src/main/scala/com/trigl/spark/main/JDBC2Mysql.scala

https://github.com/Trigl/SparkLearning/blob/master/src/main/scala/com/trigl/spark/main/HBaseDemo.scala

Refer:

http://spark.apache.org/docs/latest/sql-programming-guide.html#jdbc-to-other-databases

http://lxw1234.com/archives/2015/07/348.htm

http://blog.csdn.net/lsshlsw/article/details/48627737

以上是关于Learning Spark——Spark连接MysqlmapPartitions高效连接HBase的主要内容,如果未能解决你的问题,请参考以下文章

Learning Spark——使用spark-shell运行Word Count

Learning Spark——使用spark-shell运行Word Count

关于learning Spark翻译

Machine Learning on Spark

读learning spark lighting

Learning Spark [6] - Spark SQL高级函数