其他分享
首页 > 其他分享> > SparkCore之数据的读取与保存

SparkCore之数据的读取与保存

作者:互联网

一、文件类数据读取与保存

1.1 Text文件

object Operate_Text {

    def main(args: Array[String]): Unit = {

        //1.创建SparkConf并设置App名称
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")

        //2.创建SparkContext,该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)

        //3.1 读取输入文件
        val inputRDD: RDD[String] = sc.textFile("input/1.txt")

        //3.2 保存数据
        inputRDD.saveAsTextFile("output")

        //4.关闭连接
        sc.stop()
    }
}

1.2 Json文件

{"username": "zhangsan","age": 20}
{"username": "lisi","age": 18}
{"username": "wangwu","age": 16}
object Operate_Json {

    def main(args: Array[String]): Unit = {

        //1.创建SparkConf并设置App名称
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")

        //2.创建SparkContext,该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)

        //3.1 读取Json输入文件
        val jsonRDD: RDD[String] = sc.textFile("input/user.json")

        //3.2 导入解析Json所需的包并解析Json
        import scala.util.parsing.json.JSON
        val resultRDD: RDD[Option[Any]] = jsonRDD.map(JSON.parseFull)

        //3.3 打印结果
        resultRDD.collect().foreach(println)

        //4.关闭连接
        sc.stop()
    }
}

1.3 Sequence文件

object Operate_Sequence {

    def main(args: Array[String]): Unit = {

        //1.创建SparkConf并设置App名称
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")

        //2.创建SparkContext,该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)

        //3.1 创建rdd
        val dataRDD: RDD[(Int, Int)] = sc.makeRDD(Array((1,2),(3,4),(5,6)))

        //3.2 保存数据为SequenceFile
        dataRDD.saveAsSequenceFile("output")

        //3.3 读取SequenceFile文件
        sc.sequenceFile[Int,Int]("output").collect().foreach(println)

        //4.关闭连接
        sc.stop()
    }
}

1.4 Object对象文件

object Operate_Object {

    def main(args: Array[String]): Unit = {

        //1.创建SparkConf并设置App名称
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")

        //2.创建SparkContext,该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)

        //3.1 创建RDD
        val dataRDD: RDD[Int] = sc.makeRDD(Array(1,2,3,4))

        //3.2 保存数据
        dataRDD.saveAsObjectFile("output")

        //3.3 读取数据
        sc.objectFile[(Int)]("output").collect().foreach(println)

        //4.关闭连接
        sc.stop()
    }
}

二、文件系统类数据读取与保存

2.1 HDFS

2.2 MySQL

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>
package com.spark.day06

import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.DriverManager


/*
* 从MySQL数据库中读取数据
*
*   sc: SparkContext,   Spark程序执行的入口,上下文对象
    getConnection: () => Connection,  获取数据库连接
    sql: String,  执行SQL语句
    lowerBound: Long, 查询的起始位置
    upperBound: Long, 查询的终止位置
    numPartitions: Int, 分区数
    mapRow: (ResultSet) => T  对结果集进行处理
    *
    * jdbc连接数据库
    *   注册驱动
    *   获取连接
    *   创建数据库操作对象PrepareStatement
    *   执行SQL
    *   处理结果集
    *   关闭连接
    *
* */

object Spark04_MySQL_read {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("Test").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 创建RDD
    // 数据库连接四要素:
    var driver = "com.mysql.jdbc.Driver"
    var url = "jdbc:mysql://172.23.4.221:3306/test"
    var username = "root"
    var password = "123456"
    var sql = "select * from wyk_csdn where id >= ? and id <= ?"
    val resRDD = new JdbcRDD(
      sc,
      () => {
        // 注册驱动
        Class.forName(driver)

        // 获取连接
        DriverManager.getConnection(url, username, password)
      },
      sql,
      1,
      20,
      2,
      rs => (rs.getInt(1), rs.getString(2), rs.getString(3))
    )

    resRDD.collect().foreach(println)

    sc.stop()
  }
}
package com.spark.day06

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement}


/*
* 从MySQL数据库中写入数据
*
    *
    * jdbc连接数据库
    *   注册驱动
    *   获取连接
    *   创建数据库操作对象PrepareStatement
    *   执行SQL
    *   处理结果集
    *   关闭连接
    *
* */

object Spark05_MySQL_write {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("Test").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 创建RDD
    // 数据库连接四要素:
    var driver = "com.mysql.jdbc.Driver"
    var url = "jdbc:mysql://172.23.4.221:3306/test"
    var username = "root"
    var password = "123456"

    val rdd: RDD[(Int, String, String)] = sc.makeRDD(List((1, "banzhang", "2021-05-20 10:18:35")))

///
//  下面这段代码,需要让ps实现序列化,但是ps不是我们自己定义的类型,没有办法实现
//    //  注册驱动
//    Class.forName(driver)
//
//    // 获取连接
//    val conn: Connection = DriverManager.getConnection(url, username, password)
//
//    // 声明数据库操作的SQL语句
//    var sql:String = "insert into wyk_csdn(id, name, ins_ts) values(?, ?, ?)"
//
//    // 创建数据库操作对象
//    val ps: PreparedStatement = conn.prepareStatement(sql)
//
//
//    // 在循环体中创建连接对象,每次遍历出RDD中的一个元素,都要创建一个连接对象,效率低,不推荐使用
//    rdd.foreach{
//      case (id, name, ins_ts) => {
//
//
//        // 给参数赋值
//        ps.setInt(1, id)
//        ps.setString(2, name)
//        ps.setString(3,ins_ts)
//
//        // 执行SQL语句
//        ps.executeUpdate()
//      }
//    }
//    // 关闭连接
//    ps.close()
//    conn.close()
///

    rdd.foreachPartition{
      // datas是rdd的一个分区的数据
      datas => {
        //  注册驱动
        Class.forName(driver)

        // 获取连接
        val conn: Connection = DriverManager.getConnection(url, username, password)

        // 声明数据库操作的SQL语句
        var sql:String = "insert into wyk_csdn(id, name, ins_ts) values(?, ?, ?)"

        // 创建数据库操作对象
        val ps: PreparedStatement = conn.prepareStatement(sql)

        // 对当前分区内的数据,进行遍历
        // 这里的foreach不是算子了,是集合的方法
        datas.foreach{
          case (id, name, ins_ts) => {


            // 给参数赋值
            ps.setInt(1, id)
            ps.setString(2, name)
            ps.setString(3,ins_ts)

            // 执行SQL语句
            ps.executeUpdate()
          }
        }
        ps.close()
        conn.close()
      }
    }


    sc.stop()
  }
}

 

标签:SparkContext,ps,读取,val,SparkCore,保存,sc,SparkConf,String
来源: https://blog.csdn.net/qq_38689352/article/details/116951909