case class People(val name:String,val age:Int)  //可以聲明數(shù)據(jù)類型

object WordCount {
  def main(args:Array[String]):Unit={
    val conf = new SparkConf()
    //設(shè)置運(yùn)行模式為本地運(yùn)行，不然默認(rèn)是集群模式
    //conf.setMaster("local")  //默認(rèn)是集群模式
    //設(shè)置任務(wù)名
    conf.setAppName("WordCount").setMaster("local")
    conf.set("spark.default.parallelism","5")
    //設(shè)置SparkContext,是SparkCore的程序入口
    val sc = new SparkContext(conf)
    val Sqlsc = new SQLContext(sc)  //根據(jù)SparkContext生成SQLContext
    
    val array = Array("mark,14","kitty,23","dasi,45")
    val peopleRDD = sc.parallelize(array).map(line=>{    //生成RDD
      People(line.split(",")(0),line.split(",")(1).trim().toInt)
    })
    
    import Sqlsc.implicits._  //引入全部方法
    //將RDD轉(zhuǎn)換成DataFrame
    val df = peopleRDD.toDF()  
    //將DataFrame轉(zhuǎn)換成一個(gè)臨時(shí)的視圖
    df.createOrReplaceTempView("people")
    //使用SQL語句進(jìn)行查詢
    Sqlsc.sql("select * from people").show()
  }
}

object WordCount {
  def main(args:Array[String]):Unit={
    val conf = new SparkConf()
    //設(shè)置運(yùn)行模式為本地運(yùn)行，不然默認(rèn)是集群模式
    //conf.setMaster("local")  //默認(rèn)是集群模式
    //設(shè)置任務(wù)名
    conf.setAppName("WordCount").setMaster("local")
    conf.set("spark.default.parallelism","5")
    //設(shè)置SparkContext,是SparkCore的程序入口
    val sc = new SparkContext(conf)
    val Sqlsc = new SQLContext(sc)  //根據(jù)SparkContext生成SQLContext
    
    val array = Array("mark,14","kitty,23","dasi,45")
    //1.需要將RDD數(shù)據(jù)映射成Row,需要引入import org.apache.spark.sql.Row
    val peopleRDD = sc.parallelize(array).map(line=>{    //生成RDD
      val fields = line.split(",")
      Row(fields(0),fields(1).trim().toInt)
    })
    
    //2.創(chuàng)建StructType定義結(jié)構(gòu)
    val st:StructType = StructType(
      //字段名，字段類型，是否可以為空
      List(  //傳參是列表類型，或者使用StructField("name", StringType, true) :: StructField("age", IntegerType, true) :: Nil來構(gòu)建列表
          StructField("name",StringType,true),
          StructField("age",IntegerType,true)
          )
    )
    
    //3.使用SparkSession建立DataFrame
    val df = Sqlsc.createDataFrame(peopleRDD,st)
    //將DataFrame轉(zhuǎn)換成一個(gè)臨時(shí)的視圖
    df.createOrReplaceTempView("people")
    //使用SQL語句進(jìn)行查詢
    Sqlsc.sql("select * from people").show()
  }
}

[{"name":"dafa","age":12},{"name":"safaw","age":17},{"name":"ge","age":34}]

def main(args:Array[String]):Unit={
    val conf = new SparkConf()
    //設(shè)置運(yùn)行模式為本地運(yùn)行，不然默認(rèn)是集群模式
    //conf.setMaster("local")  //默認(rèn)是集群模式
    //設(shè)置任務(wù)名
    conf.setAppName("WordCount").setMaster("local")
    //設(shè)置SparkContext,是SparkCore的程序入口
    val sc = new SparkContext(conf)
    val Sqlsc = new SQLContext(sc)  //根據(jù)SparkContext生成SQLContext
    
    //通過json數(shù)據(jù)直接創(chuàng)建DataFrame
    val df = Sqlsc.read.json("E:\\1.json")
    
    //將DataFrame轉(zhuǎn)換成一個(gè)臨時(shí)的視圖
    df.createOrReplaceTempView("people1")
    //使用SQL語句進(jìn)行查詢
    Sqlsc.sql("select * from people1").show()
  }

val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    //方式一
    val df1 = sqlContext.read.json("E:\\666\\people.json")
    val df2 = sqlContext.read.parquet("E:\\666\\users.parquet")
    //方式二
    val df3 = sqlContext.read.format("json").load("E:\\666\\people.json")
    val df4 = sqlContext.read.format("parquet").load("E:\\666\\users.parquet")
    //方式三，默認(rèn)是parquet格式
    val df5 = sqlContext.load("E:\\666\\users.parquet")
　  //方式四，使用MySQL進(jìn)行數(shù)據(jù)源讀取
    val url = "jdbc:mysql://192.168.123.102:3306/hivedb"
    val table = "dbs"
    val properties = new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","root")
    //需要傳入Mysql的URL、表明、properties（連接數(shù)據(jù)庫的用戶名密碼）
    val df = sqlContext.read.jdbc(url,table,properties)
    df.createOrReplaceTempView("dbs")
    sqlContext.sql("select * from dbs").show()

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>

<configuration>
        <property>
                <name>javax.jdo.option.ConnectionURL</name>
                <value>jdbc:mysql://localhost:3306/hivedb?createDatabaseIfNotExist=true</value>
                <description>JDBC connect string for a JDBC metastore</description>
                <!-- 如果 mysql 和 hive 在同一個(gè)服務(wù)器節(jié)點(diǎn)，那么請(qǐng)更改 hadoop02 為 localhost -->
        </property>
        <property>
                <name>javax.jdo.option.ConnectionDriverName</name>
                <value>com.mysql.jdbc.Driver</value>
                <description>Driver class name for a JDBC metastore</description>
        </property>
        <property>
                <name>javax.jdo.option.ConnectionUserName</name>
                <value>root</value>
                <description>username to use against metastore database</description>
        </property>
        <property>
                <name>javax.jdo.option.ConnectionPassword</name>
                <value>root</value>
        <description>password to use against metastore database</description>
        </property>
    <property>
                <name>hive.metastore.warehouse.dir</name>
                <value>/hive/warehouse</value>
                <description>hive default warehouse, if nessecory, change it</description>
        </property>  
</configuration>

hive-site.xml配置文件

val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    sqlContext.sql("select * from myhive.student").show() 

val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val df1 = sqlContext.read.json("E:\\666\\people.json")
    //方式一
    df1.write.json("E:\\111")
    df1.write.parquet("E:\\222")
    //方式二
    df1.write.format("json").save("E:\\333")
    df1.write.format("parquet").save("E:\\444")
    //方式三
    df1.write.save("E:\\555")

df1.write.format("parquet").mode(SaveMode.Ignore).save("E:\\444")

(1,"Tom")  (2,"Mary")