以编程方式执行Spark SQL查询的两种实现方式

天策

发布于 2018-06-22 06:53:34

2.3K00

代码可运行

文章被收录于专栏：行者悟空行者悟空

运行总次数：0

代码可运行

摘要

在自定义的程序中编写Spark SQL查询程序

1.通过反射推断Schema

package com.itunic.sql

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by itunic.com on 2017/1/2.
  * Spark SQL
  * 通过反射推断Schema
  * by me:
  * 我本沉默是关注互联网以及分享IT相关工作经验的博客，
  * 主要涵盖了操作系统运维、计算机编程、项目开发以及系统架构等经验。
  * 博客宗旨：把最实用的经验，分享给最需要的你，
  * 希望每一位来访的朋友都能有所收获！
  *
  */
object InferringSchema {
  def main(args: Array[String]): Unit = {
 //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("InferringSchema").setMaster("local")
 //SQLContext要依赖SparkContext
    val sc = new SparkContext(conf)
 //创建SQLContext
    val sqlContext = new SQLContext(sc)
 //从指定的地址创建RDD
    val lineRdd = sc.textFile("F:\\test\\input\\wc.txt").map(f => {
      val fields = f.split("\t")
 //将RDD和case class关联
      Person(fields(0).toLong, fields(1), fields(2).toInt)
    })
 //导入隐式转换，如果不导入无法将RDD转换成DataFrame
 //将RDD转换成DataFrame
 import sqlContext.implicits._
    val personDF = lineRdd.toDF
 //注册表
    personDF.registerTempTable("t_person")
 //传入SQL
    val df = sqlContext.sql("select * from t_person order by age desc limit 2")
 //显示
    df.show()
 //以json方式写入hdfs
 //df.write.json("hdfs://ns1:9000/wc")
    sc.stop()
  }
}

//定义样例类
case class Person(id: Long, userName: String, age: Int)

2.通过StructType直接指定Schema

package com.itunic.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types._

/**
  * Created by itunic.com on 2017/1/2.
  *  Spark SQL
  * 通过StructType直接指定Schema
  * by me:
  * 我本沉默是关注互联网以及分享IT相关工作经验的博客，
  * 主要涵盖了操作系统运维、计算机编程、项目开发以及系统架构等经验。
  * 博客宗旨：把最实用的经验，分享给最需要的你，
  * 希望每一位来访的朋友都能有所收获！
  *
  */
object SpecifyingSchema {
  def main(args: Array[String]): Unit = {
 //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("SpecifyingSchema").setMaster("local")
 //SQLContext要依赖SparkContext
    val sc = new SparkContext(conf)
 //创建SQLContext
    val sqlContext = new SQLContext(sc)
 //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        StructField("id", LongType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )

    val lineRdd = sc.textFile("F:\\test\\input\\wc.txt").map(f => {
      val fields = f.split("\t")
      Row(fields(0).toLong, fields(1), fields(2).toInt)
    })
 //将schema信息应用到lineRdd上
    val personDF = sqlContext.createDataFrame(lineRdd, schema)
    personDF.registerTempTable("t_person")
 //传入SQL
    val df = sqlContext.sql("select * from t_person order by age desc limit 2")
 //显示
    df.show()
 //以json方式写入hdfs
 //df.write.json("hdfs://ns1:9000/wc")
    sc.stop()
  }
}