Spark 数据读取与保存（输入、输出）

scala> val result = json.map(JSON.parseFull)
result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[10] at map at <console>:27

scala> result.collect
res11: Array[Option[Any]] = Array(Some(Map(name -> Michael)), Some(Map(name -> Andy, age -> 30.0)), Some(Map(name -> Justin, age -> 19.0)))

4.1.3 Sequence 文件 

[lxl@hadoop102 seqFile]$ pwd
/opt/module/spark/seqFile

[lxl@hadoop102 seqFile]$ ll
总用量 8
-rw-r--r-- 1 atguigu atguigu 108 10 月 9 10:29 part-00000
-rw-r--r-- 1 atguigu atguigu 124 10 月 9 10:29 part-00001
-rw-r--r-- 1 atguigu atguigu 0 10 月 9 10:29 _SUCCESS

[lxl@hadoop102 seqFile]$ cat part-00000
SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritableط

def main(args: Array[String]) {
  val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp")
  val sc = new SparkContext(sparkConf)
  val data = sc.parallelize(List("Female", "Male","Female"))
  data.foreachPartition(insertData)
}

def insertData(iterator: Iterator[String]): Unit = {
　　Class.forName ("com.mysql.jdbc.Driver").newInstance()
　　val conn = java.sql.DriverManager.getConnection("jdbc:mysql://master01:3306/rdd", "root","hive")

　　iterator.foreach(data => {
　　　　val ps = conn.prepareStatement("insert into rddtable(name) values (?)")
　　　　ps.setString(1, data)
　　　　ps.executeUpdate()
　　})
} 

[lxl@hadoop102 spark]$ cp /opt/module/hive/lib/mysql-connector-java-5.1.27-bin.jar ./jars/

scala> val rdd = new org.apache.spark.rdd.JdbcRDD(sc, () => {
     |       Class.forName("com.mysql.jdbc.Driver")
     |       java.sql.DriverManager.getConnection("jdbc:mysql://hadoop102:3306/rdd", "root", "000000") 
     |     },
     |       "select * from `rddtable` where id >= ? and id <= ?;",
     |       1,
     |       10,
     |       1,
     |       r => (r.getInt(1), r.getString(2))
     |     )
rdd: org.apache.spark.rdd.JdbcRDD[(Int, String)] = JdbcRDD[1] at JdbcRDD at <console>:24

scala> println(rdd.count())
3

scala> rdd.foreach(println)
(1,zhangsan)
(2,lisi)
(3,wangwu)

package com.lxl

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.hbase.util.Bytes
object HBaseSpark {
  def main(args: Array[String]): Unit = {
//创建 spark 配置信息
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JdbcRDD")
//创建 SparkContext
    val sc = new SparkContext(sparkConf)
//构建 HBase 配置信息
    val conf: Configuration = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104")
    conf.set(TableInputFormat.INPUT_TABLE, "rddtable")
//从 HBase 读取数据形成 RDD
    val hbaseRDD: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result])

    val count: Long = hbaseRDD.count()
    println(count)
//对 hbaseRDD 进行处理
    hbaseRDD.foreach {
      case (_, result) =>
        val key: String = Bytes.toString(result.getRow)
        val name: String = Bytes.toString(result.getValue(Bytes.toBytes("info"),Bytes.toBytes("name")))
        val color: String = Bytes.toString(result.getValue(Bytes.toBytes("info"),Bytes.toBytes("color")))
        println("RowKey:" + key + ",Name:" + name + ",Color:" + color)
    }
//关闭连接
    sc.stop()
  }
}