Spark连接MySQL

object ConnectMysql {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local[4]").appName(this.getClass.getName).getOrCreate()
    //设置要访问的mysql的url,表名
    val url = "jdbc:mysql://singer:3306/kb10"
    val tableName ="hive_shop"
    val props=new Properties()
    //设置要访问的mysql的用户名,密码,Drive
    props.setProperty("user","root")
    props.setProperty("password","kb10")
    props.setProperty("driver","com.mysql.jdbc.Driver")
    //通过spark. read.jdbc方法读取mysql中数据
    val df: DataFrame = spark.read.jdbc(url,tableName,props)
    df.show()
    
    
    //将DataFrame数据写入到MySQL中,追加方式
//    df.write.mode("append").jdbc(url,tableName,props)

spark和MySQL中运行结果一致：

Spark连接Hive

object ConnectHive {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local[2]")
      .enableHiveSupport()
      .config("hive.metastore.uris", "thrift://192.168.181.129:9083")
      .appName(this.getClass.getName).getOrCreate()

    val df: DataFrame = spark.sql("show databases")
    df.show()
  }
}

spark和Hive的运行结构截图一致：

Spark连接Hbase

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.SparkSession


object ConnectHbase {
  def main(args: Array[String]): Unit = {
    val conf = HBaseConfiguration.create()

    conf.set("hbase.zookeeper.quorum","192.168.181.129")
    conf.set("hbase.zookeeper.property.clientPort","2181")
    conf.set(TableInputFormat.INPUT_TABLE,"kb10:customer")

    val spark = SparkSession.builder().appName("HBaseTest")
      .master("local[2]")
      .getOrCreate()
    val sc= spark.sparkContext


    val rdd1= sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result]
    ).cache()

    println("count="+rdd1.count())
    import spark.implicits._
    //遍历输出
    rdd1.foreach({case (_,result) =>
      //通过result.getRow来获取行键
      val key = Bytes.toString(result.getRow)
      //通过result.getValue("列簇","列名")来获取值
      //需要使用getBytes将字符流转化为字节流
      val city = Bytes.toString(result.getValue("addr".getBytes,"city".getBytes))
      val country = Bytes.toString(result.getValue("addr".getBytes,"country".getBytes))
      val age = Bytes.toString(result.getValue("order".getBytes,"age".getBytes))
      
      println("Row key:"+key+" city:"+city+" country:"+country+" age:"+age)
    })

  }
}