Spark-SQL的具体编程场景

Posted 2021-02-01

入门案例：

object SparkSqlTest {
    def main(args: Array[String]): Unit = {
        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        /**
          * 注意在spark 2.0之后：
          * val sqlContext = new SQLContext(sparkContext)
          * val hiveContext = new HiveContext(sparkContext)
          * 主构造器被私有化，所以这里只能使用SparkSession对象创建
          */
        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //加载数据为DataFrame，这里加载的是json数据
        //数据格式：{name:‘‘,age:18}
        val perDF: DataFrame = sqlContext.read.json("hdfs://zzy/data/person.json")

        //查看二维表结构
        perDF.printSchema()

        //查看数据，默认显示20条记录
        perDF.show()

        //复杂查询
        perDF.select("name").show() //指定字段进行查询
        perDF.select(new Column("name"),new Column("age").>(18)).show()  //指定查询条件进行查询
        perDF.select("name","age").where(new Column("age").>(18)).show() //指定查询条件进行查询
        perDF.select("age").groupBy("age").avg("age") //聚合操作
    }
}

如果对入门案例不太了解的话，接下来分步骤的介绍：

（1）RDD/DataSet//DataFrame/list 之间的转化

   通过RDD转换为DataFrame/DataSet，有两种方式：
    - 通过反射的方式将RDD或者外部的集合转化为dataframe/datasets
    - 要通过编程动态的来将外部的集合或者RDD转化为dataframe或者dataset
   注意：如果是dataFrame对应的是java bean ，如果是dataSet对应的是case class
 

通过反射的方式将RDD或者外部的集合转化为dataframe/datasets

数据准备：

case class Student(name:String, birthday:String, province:String)
val stuList = List(
      new Student("委xx", "1998-11-11", "山西"),
      new Student("吴xx", "1999-06-08", "河南"),
      new Student("戚xx", "2000-03-08", "山东"),
      new Student("王xx", "1997-07-09", "安徽"),
      new Student("薛xx", "2002-08-09", "辽宁")
    )

list --> DataFrame:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext

        /**
          * list--->DataFrame
          * 将scala集合转换为java集合
          */
        val javaList: util.List[Student] = JavaConversions.seqAsJavaList(stuList)
        val stuDF: DataFrame = sqlContext.createDataFrame(javaList,classOf[Student])
        val count = stuDF.count()
        println(count)

RDD --> DataFrame:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext
        /**
          * RDD--->DataFrame
          */
        val stuRDD: RDD[Student] = sc.makeRDD(stuList)
        val stuDF: DataFrame = sqlContext.createDataFrame(stuRDD,classOf[Student])
        val count = stuDF.count()
        println(count)

list --> DataSet:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext
        /**
          * list--->DataSet
          */
        //如果创建Dataset 必须导入下面的隐式转换
        import spark.implicits._
        val stuDF: Dataset[Student] = sqlContext.createDataset(stuList)
        stuDF.createTempView("student")
        //使用完整的sql语句进行查询，使用反射的方式，只有Dataset可以，dataFrame不行
        val sql=
            """
              |select * from student
            """.stripMargin
        spark.sql(sql).show()

RDD --> DataSet:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext
        /**
          * RDD--->DataSet
          */
        //如果创建Dataset 必须导入下面的隐式转换
        import spark.implicits._
        val stuRDD: RDD[Student] = sc.makeRDD(stuList)
        val stuDF: Dataset[Student] = sqlContext.createDataset(stuRDD)
        stuDF.createTempView("student")
        //使用完整的sql语句进行查询，使用反射的方式，只有Dataset可以，dataFrame不行
        val sql=
            """
              |select * from student
            """.stripMargin
        spark.sql(sql).show()

通过编程动态的来将外部的集合或者RDD转化为dataframe或者dataset

list --> DataFrame:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext

        //list-DataFrame

        //1.将list中的元素全部转化为Row
        val RowList: List[Row] = stuList.map(item => {
            Row(item.name, item.birthday, item.province)
        })
        //2.构建元数据
        val schema=StructType(List(
            StructField("name",DataTypes.StringType),
            StructField("birthday",DataTypes.StringType),
            StructField("province",DataTypes.StringType)
        ))
        //将scala的集合转化为java集合
        val javaList = JavaConversions.seqAsJavaList(RowList)
        val stuDF = spark.createDataFrame(javaList,schema)
        stuDF.createTempView("student")
        //使用完整的sql语句进行查询，使用动态编程的方式，Dataset、dataFrame都可以
        val sql=
            """
              |select * from student
            """.stripMargin
        spark.sql(sql).show()

RDD--> DataFrame:

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
            .setMaster("local[2]")
            .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(Array(classOf[Student]))
        val spark: SparkSession = SparkSession.builder().config(conf)
            .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext

        //RDD-DataFrame

        //将RDD中的元素转换为Row
        val RowRDD: RDD[Row] = sc.makeRDD(stuList).map(item => {
            Row(item.name, item.birthday, item.province)
        })

        //2.构建元数据
        val schema=StructType(List(
            StructField("name",DataTypes.StringType),
            StructField("birthday",DataTypes.StringType),
            StructField("province",DataTypes.StringType)
        ))
        val stuDF = spark.createDataFrame(RowRDD,schema)
        stuDF.createTempView("student")
        //使用完整的sql语句进行查询，使用动态编程的方式，Dataset、dataFrame都可以
        val sql=
            """
              |select * from student
            """.stripMargin
        spark.sql(sql).show()

由于构建DataFrame和构建DataSet一模一样，这里就不在演示
 

（2）spark SQL加载数据的方式

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
                .setMaster("local[2]")

        val spark: SparkSession = SparkSession.builder().config(conf)
                .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext

        //早期版本加载：parquet文件
        sqlContext.load("hdfs://zzy/hello.parquet")
        //加载json数据
        sqlContext.read.json("hdfs://zzy/hello.json")
        //加载普通文件
        sqlContext.read.text("hdfs://zzy/hello.txt")
        //加载csv
        sqlContext.read.csv("hdfs://zy/hello.csv")
        //读取jdbc的数据
        val url="jdbc:mysql://localhost:3306/hello"
        val properties=new Properties()
        properties.setProperty("user","root")
        properties.setProperty("password","123456")
        val tableName="book"
        sqlContext.read.jdbc(url,tableName,properties)

（3）spark SQL数据落地的方式

        //屏蔽多余的日志
        Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        Logger.getLogger("org.project-spark").setLevel(Level.WARN)
        //构建编程入口
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparkSqlTest")
                .setMaster("local[2]")

        val spark: SparkSession = SparkSession.builder().config(conf)
                .getOrCreate()

        //创建sqlcontext对象
        val sqlContext: SQLContext = spark.sqlContext
        //创建sparkContext
        val sc: SparkContext = spark.sparkContext
        val testFD: DataFrame = sqlContext.read.text("hdfs://zzy/hello.txt")

        //写入到普通文件
        testFD.write.format("json") //以什么格式写入
                .mode(SaveMode.Append)  //写入方式
                .save("hdfs://zzy/hello.json")  //写入的文件位置

        //写入到数据库
        val url="jdbc:mysql://localhost:3306/hello"
        val table_name="book"
        val prots=new Properties()
        prots.put("user","root")
        prots.put("password","123456")
        testFD.write.mode(SaveMode.Append).jdbc(url,table_name,prots)