读取hive库数据

pom.xml依赖配置

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-hive_2.11</artifactId>
  <version>2.1.1</version>
</dependency>

读取hive数据demo

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
object Main {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
              .setMaster("spark://master:7077")//申明spark运行模式
              .setAppName("risk")//设置job名称(可不写)
    val spark = SparkSession.builder()//spark-2.0采用SparkSession代替sparkContext
                .config(conf)
                .enableHiveSupport()//添加对HIVE的支持，否则无法访问hive库
                .getOrCreate()
    import spark.implicits._
    spark.sql("use bmkp")
    val df= spark.sql("select * from customer")//在hive中执行sql语句，返回DataSet格式数据
    df.show()
    spark.stop()
  }
}

---

读取mysql数据

pom.xml配置文件

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-hive_2.11</artifactId>
  <version>2.1.1</version>
</dependency>

读取mysql数据demo

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
object Main {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
                  .setMaster("spark://master:7077")//申明spark运行模式
                  .setAppName("risk")//设置job名称(可不写)
    val spark = SparkSession.builder()//spark-2.0采用SparkSession代替sparkContext
                 .config(conf)
                 .enableHiveSupport()//添加对HIVE的支持，否则无法访问hive库
                 .getOrCreate()
//读取mysql中数据，返回数据类型为DataSet
val df = spark.read
        .format("jdbc")
        .options(Map("url" ->   
//配置mysql连接参数，包括mysql ip 端口  数据库名称 登录名和密码
"jdbc:mysql://***.***.***.***:3036/bmkpstress?user=root&password=**********",
//定义驱动程序
         "driver"->"com.mysql.jdbc.Driver",
//编写sql  在mysql中执行该sql并返回数据
         "dbtable" -> "(select * from test group by id) as aaa"))
          .load()
    spark.stop()
  }
}

---

SPARKSTREAMING读取kafka数据

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.1.1</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-hive_2.11</artifactId>
  <version>2.1.1</version>
</dependency>

读取kafka数据demo

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

object Main {
  def main(args:Array[String]):Unit={
    val conf = new SparkConf().setMaster("spark://master:7077")
      .setAppName("kafka_hive");
    val spark = SparkSession.builder().master("spark://master:7077").config(conf).enableHiveSupport().getOrCreate()
    var ssc = new StreamingContext(conf, Seconds(10));
    var topics = Array("service_cksc","service_ckxc","service_dcyy");//kafka  topic名称
    var group = "bmkp" //定义groupID
    val kafkaParam = Map(   //申明kafka相关配置参数
      "bootstrap.servers" -> "***.104.42.127:19092,***.104.202.222:19092,***.135.73.152:19092", //kafka 集群IP及端口
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> group, //定义groupID
      "auto.offset.reset" -> "earliest",//设置丢数据模式  有 earliest,latest, none
      "enable.auto.commit" -> (false: java.lang.Boolean)//设置是否自动存储offset 这里设置为否
    );
    val offsetRanges = Array()
    var stream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParam))//从kafka读取数据 获取数据流
    stream.foreachRDD { rdd =>
      import spark.implicits._
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //获取offset
      /*
      这里处理从kafka获取的数据，在确定获取的数据已经存储或者处理后将该RDD的offset存储
       */
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) //存储offset
    }
  }
}

---

SPARK写数据到HIVE

pom.xml配置信息

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-hive_2.11</artifactId>
  <version>2.1.1</version>
</dependency

写数据到hive库demo

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object Main {
  case class Person(name:String,col1:Int,col2:String)
  def main(args:Array[String]):Unit={
    val conf = new SparkConf()
      .setMaster("spark://master:7077")//申明spark运行模式
      .setAppName("kettle")//设置job名称(可不写)
    val spark = SparkSession.builder()//spark-2.0采用SparkSession代替sparkContext
      .config(conf)
      .enableHiveSupport()//添加对HIVE的支持，否则无法访问hive库
      .getOrCreate()
    import spark.implicits._ //引入隐式转换 否则RDD无法转换成DataSet(DataFrame)
    spark.sql("use DataBaseName"//在hive中执行sql语句
    val data = spark.read.textFile("path")//读取hdfs中的文件，返回的是RDD格式数据，RDD格式数据不能直接写入hive,（这里代表任意的RDD类型数据）
      .map(x=>x.split("，"))
      .map(x=>Person(x(0),x(1).toInt,x(2)))//利用用例类将RDD格式居转换成DataSetG格式数据，从而可以写入hive中
    data.toDF().createOrReplaceTempView("table1") //将DataSet格式数据映射到临时表中
    spark.sql("insert into table2 partition(date='2015-04-02') select name,col1,col2 from table1")//在hive上运行sql语句将临时表中数据抽出并存入hive中
    spark.close()
  }
}

---

写数据到mysql

pom.xml配置

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.1.1</version>
</dependency>

spark 写数据到mysql库demo 1

  import java.sql.{Connection, DriverManager, PreparedStatement}
  import org.apache.spark.SparkConf
  import org.apache.spark.sql.SparkSession
object Main {
  case class Blog(name: String, count: Int)
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setMaster("spark://master:7077")//申明spark运行模式
      .setAppName("kettle")//设置job名称(可不写)
    val spark = SparkSession.builder()//spark-2.0采用SparkSession代替sparkContext
      .config(conf)
      .enableHiveSupport()//添加对HIVE的支持，否则无法访问hive库
      .getOrCreate()
    //获取RDD数据  这里只是做一个实例 代表spark处理产生的所有RDD类型的数据
    val data = spark.sparkContext.parallelize(List(("www", 10), ("iteblog", 20), ("com", 30)))
    var conn: Connection = null//定义mysql连接
    var ps: PreparedStatement = null
    val sql = "insert into blog(name, count) values (?, ?)"//需要执行的sql语句，两个 “？”代表后面需要替换的数据
    data.foreachPartition(rdd=>
      try {
        //具体定义mysql的驱动管理器，主要设置mysql地址   端口  数据库  用户名  密码
        conn = DriverManager.getConnection("jdbc:mysql://***.***.***.***:3306/test","root", "******")
        rdd.toIterator.foreach(data => {
          ps = conn.prepareStatement(sql)
          ps.setString(1, data._1)//将需要写入mysql的数据进行映射
          ps.setInt(2, data._2)
          ps.executeUpdate()//在mysql上执行sql语句将数据插入到相应的表中
        })
      } catch {
        case e: Exception => println("Mysql Exception")
      } finally {
        if (ps != null) {
          ps.close()
        }
        if (conn != null) {
          conn.close()//关闭mysql连接
        }
      })
  }
}

写数据到mysql库demo 2

  import org.apache.spark.SparkConf
  import org.apache.spark.sql.{SaveMode, SparkSession}

object Main {
  case class Blog(name: String, count: Int)
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setMaster("spark://master:7077") //申明spark运行模式
      .setAppName("kettle")
    //设置job名称(可不写)
    val spark = SparkSession.builder() //spark-2.0采用SparkSession代替sparkContext
      .config(conf)
      .enableHiveSupport() //添加对HIVE的支持，否则无法访问hive库
      .getOrCreate()
    //获取RDD数据  这里只是做一个实例 代表spark处理产生的所有RDD类型的数据
    val data = spark.sparkContext.parallelize(List(("www", 10), ("iteblog", 20), ("com", 30)))
    import spark.implicits._
    val df = data.map(x=>new Blog(x._1,x._2)).toDF()//将RDD类型数据转换成DataSet类型
    df.write.mode(SaveMode.Append).format("jdbc")
      .option("url", "jdbc:mysql://***.***.***.***:3306/test")//定义mysql 地址 端口 数据库
      .option("dbtable", "blog")//定义需要插入的mysql目标表
      .option("user", "****")//定义登录用户名
      .option("password", "************")//定义登录密码
      .save()//保存数据
  }
}