宝贝腿开大点我添添你视频男男,中文字幕熟女人妻av一区二区三区,爱色成人网,大地资源高清播放在线观看在线电影在线观看 ,777米奇影视第四色

集團(tuán)站切換校區(qū)

驗證碼已發(fā)送,請查收短信

復(fù)制成功
微信號:togogoi
添加微信好友, 詳細(xì)了解課程
已復(fù)制成功,如果自動跳轉(zhuǎn)微信失敗,請前往微信添加好友
打開微信
圖標(biāo)

學(xué)習(xí)文章

當(dāng)前位置:首頁 > >學(xué)習(xí)文章 > >

{大數(shù)據(jù)}Spark SQL

發(fā)布時間: 2018-01-12 09:29:00

        我們已經(jīng)學(xué)習(xí)了Hive,它是將Hive SQL轉(zhuǎn)換成MapReduce然后提交到集群上執(zhí)行,大大簡化了編寫MapReduce的程序的復(fù)雜性,由于MapReduce這種計算模型執(zhí)行效率比較慢。所以Spark SQL的應(yīng)運(yùn)而生,它是將Spark SQL轉(zhuǎn)換成RDD,然后提交到集群執(zhí)行,執(zhí)行效率非???!?

        1.易整合:

            ?

          2.統(tǒng)一的數(shù)據(jù)訪問方式:

            

        3.兼容Hive?:

            

        4.標(biāo)準(zhǔn)的數(shù)據(jù)連接?:

            

DataFrames:

        與RDD類似,DataFrame也是一個分布式數(shù)據(jù)容器。然而DataFrame更像傳統(tǒng)數(shù)據(jù)庫的二維表格,除了數(shù)據(jù)以外,還記錄數(shù)據(jù)的結(jié)構(gòu)信息,即schema。同時,與Hive類似,DataFrame也支持嵌套數(shù)據(jù)類型(struct、array和map)。從API易用性的角度上 看,DataFrame API提供的是一套高層的關(guān)系操作,比函數(shù)式的RDD API要更加友好,門檻更低。由于與R和Pandas的DataFrame類似,Spark DataFrame很好地繼承了傳統(tǒng)單機(jī)數(shù)據(jù)分析的開發(fā)體驗。??

        

創(chuàng)建DataFrames:

         在Spark SQL中SQLContext是創(chuàng)建DataFrames和執(zhí)行SQL的入口,在spark-2.2.0中已經(jīng)內(nèi)置了一個spark:

        1.在本地創(chuàng)建一個文件,有三列,分別是id、name、age,用空格分隔,然后上傳到hdfs上

            [hadoop@hdp08 ~]$ hadoop fs -rm -r /work/person.txt

                person.txt                 

                    1,stone,30

                    2,jacky,28

                    3,mary,20

                    4,micky,27

                    5.Tom,32

        2.在spark shell執(zhí)行下面命令,讀取數(shù)據(jù),將每一行的數(shù)據(jù)使用列分隔符分割

            scala>val lineRDD = sc.textFile("hdfs://hdp08:9000/work/person.txt").map(_.split(","))

        3.定義case class(相當(dāng)于表的schema)?

            scala>case class Person(id:Int, name:String, age:Int)?

        4.將RDD和case class關(guān)聯(lián)

            scala>val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))

        5.將RDD轉(zhuǎn)換成DataFrame

            scala>val personDF = personRDD.toDF

        6.對DataFrame進(jìn)行處理

            scala>personDF.show

            

DataFrame常用操作?

?        //查看DataFrame中的內(nèi)容

        personDF.show


        //查看DataFrame部分列中的內(nèi)容

        personDF.select(personDF.col("name")).show

        personDF.select(col("name"), col("age")).show

        personDF.select("name").show


        //打印DataFrame的Schema信息

        personDF.printSchema


        //查詢所有的name和age,并將age+1

        personDF.select(col("id"), col("name"), col("age") + 1).show

        personDF.select(personDF("id"), personDF("name"), personDF("age") + 1).show

        

        

        //過濾age大于等于18的

        personDF.filter(col("age") >= 18).show

        

?        

        //按年齡進(jìn)行分組并統(tǒng)計相同年齡的人數(shù)

        personDF.groupBy("age").count().show()

        

SQL風(fēng)格語法:

        如果想使用SQL風(fēng)格的語法,需要將DataFrame注冊成表

        personDF.createOrReplaceTempView("t_person")

        val sqlDF = spark.sql("SELECT * FROM t_person")

        sqlDF.show()

        

        //查詢年齡較大的前兩名

        spark.sql("select * from t_person order by age desc limit 2").show

        

        //顯示表的Schema信息

        spark.sql("desc t_person").show

        

        

以編程方式執(zhí)行Spark SQL查詢:

        編寫Spark SQL查詢程序

        通過反射推斷Schema

        創(chuàng)建一個object為net.togogo.sql.InferringSchema?         

package net.togogo.sql

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

import org.apache.spark.sql.SQLContext

import org.apache.spark.sql.SparkSession


object InferringSchema {

 def main(args: Array[String]) {

//    val spark = SparkSession

//      .builder()

//      .appName("Spark SQL basic example")

//      .config("spark.some.config.option", "some-value")

//      .getOrCreate();


   //創(chuàng)建SparkConf()并設(shè)置App名稱

   val conf = new SparkConf().setAppName("SQL-1");

   //SQLContext要依賴SparkContext

   val sc = new SparkContext(conf);

   //創(chuàng)建SQLContext

   val sqlContext = new SQLContext(sc);


   //從指定的地址創(chuàng)建RDD

   val lineRDD = sc.textFile(args(0)).map(_.split(","));


   //創(chuàng)建case class

   //將RDD和case class關(guān)聯(lián)

   val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt));

   //導(dǎo)入隱式轉(zhuǎn)換,如果不到人無法將RDD轉(zhuǎn)換成DataFrame

   //將RDD轉(zhuǎn)換成DataFrame

   import sqlContext.implicits._

   val personDF = personRDD.toDF();

   //注冊表

   personDF.createOrReplaceTempView("t_person");

   //傳入SQL

   val df = sqlContext.sql("select * from t_person order by age desc limit 2");

   //將結(jié)果以JSON的方式存儲到指定位置

   df.write.json(args(1));

   //停止Spark Context

   sc.stop();

 }

}

//case class一定要放到外面

case class Person(id: Int, name: String, age: Int)


將程序打成jar包,上傳到spark集群,提交Spark任務(wù)

/home/hadoop/apps/spark/bin/spark-submit \

--class net.togogo.sql.InferringSchema \

--master spark://hdp08:7077 \

/home/hadoop/schema.jar \

hdfs://hdp08:9000/work/person.txt \

hdfs://hdp08:9000/work/out


查看運(yùn)行結(jié)果

[hadoop@hdp08 ~]$ hadoop fs -cat /work/out/part-00000-af7ccf43-af95-48f1-8470-e8d309f8725d-c000.json


通過StructType直接指定Schema:

創(chuàng)建一個object為net.togogo.sql.SpecifyingSchema:

        

package net.togogo.sql;

import org.apache.spark.sql.{Row, SQLContext}

import org.apache.spark.sql.types._

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


object SpecifyingSchema {

 def main(args: Array[String]) {

   //創(chuàng)建SparkConf()并設(shè)置App名稱

   val conf = new SparkConf().setAppName("SQL-2")

   //SQLContext要依賴SparkContext

   val sc = new SparkContext(conf)

   //創(chuàng)建SQLContext

   val sqlContext = new SQLContext(sc)

   //從指定的地址創(chuàng)建RDD

   val personRDD = sc.textFile(args(0)).map(_.split(" "))

   //通過StructType直接指定每個字段的schema

   val schema = StructType(

     List(

       StructField("id", IntegerType, true),

       StructField("name", StringType, true),

       StructField("age", IntegerType, true)

     )

   )

   //將RDD映射到rowRDD

   val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))

   //將schema信息應(yīng)用到rowRDD上

   val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)

   //注冊表

   personDataFrame.registerTempTable("t_person")

   //執(zhí)行SQL

   val df = sqlContext.sql("select * from t_person order by age desc limit 4")

   //將結(jié)果以JSON的方式存儲到指定位置

   df.write.json(args(1))

   //停止Spark Context

   sc.stop()

 }

}


將程序打成jar包,上傳到spark集群,提交Spark任務(wù)

/home/hadoop/apps/spark/bin/spark-submit \

--class net.togogo.sql.InferringSchema \

--master spark://hdp08:7077 \

/home/hadoop/schema.jar \

hdfs://hdp08:9000/work/person.txt \

hdfs://hdp08:9000/work/out1

查看結(jié)果

[hadoop@hdp08 ~]$ hadoop fs -cat /work/out1/part-00000-af7ccf43-af95-48f1-8470-e8d309f8725d-c000.json


數(shù)據(jù)源:

    JDBC:

        Spark SQL可以通過JDBC從關(guān)系型數(shù)據(jù)庫中讀取數(shù)據(jù)的方式創(chuàng)建DataFrame,通過對DataFrame一系列的計算后,還可以將數(shù)據(jù)再寫回關(guān)系型數(shù)據(jù)庫中。

        從MySQL中加載數(shù)據(jù)(Spark Shell方式)?:

1.啟動Spark Shell,必須指定mysql連接驅(qū)動jar包

/home/hadoop/apps/spark/bin/spark-shell  \

--master spark://hdp08:7077 \

--jars /home/hadoop/mysql-connector-java-5.1.45.jar \

--driver-class-path  /home/hadoop/mysql-connector-java-5.1.45.jar

--executor-memory 1g

--total-executor-cores 2

2.從mysql中加載數(shù)據(jù)

scala> case class Emp(empno: Int, ename: String, job:String,mgr:Int,hiredate:java.util.Date,sal:Float,comm:Float,deptno:Int)

scala>var sqlContext = new org.apache.spark.sql.SQLContext(sc);

scala> val jdbcDF = sqlContext.read.format("jdbc").options(Map("url" -> "jdbc:mysql://hdp08:3306/sqoopdb", "driver" -> "com.mysql.jdbc.Driver", "dbtable" -> "emp", "user" -> "root", "password" -> "root")).load()


3.執(zhí)行查詢

jdbcDF.show()

?

將數(shù)據(jù)寫入到MySQL中(打jar包方式)

1.編寫Spark SQL程序?:

package net.togogo.sql


import java.util.Properties

import org.apache.spark.sql.{SQLContext, Row}

import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}

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


object JdbcRDD {

 def main(args: Array[String]) {

   val conf = new SparkConf().setAppName("MySQL-Demo")

   val sc = new SparkContext(conf)

   val sqlContext = new SQLContext(sc)

   //通過并行化創(chuàng)建RDD

   val personRDD = sc.parallelize(Array("1 tom 5", "2 jerry 3", "3 kitty 6")).map(_.split(" "))

   //通過StructType直接指定每個字段的schema

   val schema = StructType(

     List(

       StructField("id", IntegerType, true),

       StructField("name", StringType, true),

       StructField("age", IntegerType, true)

     )

   )

   //將RDD映射到rowRDD

   val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))

   //將schema信息應(yīng)用到rowRDD上

   val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)

   //創(chuàng)建Properties存儲數(shù)據(jù)庫相關(guān)屬性

   val prop = new Properties()

   prop.put("user", "root")

   prop.put("password", "root")

   //將數(shù)據(jù)追加到數(shù)據(jù)庫

   personDataFrame.write.mode("append").jdbc("jdbc:mysql://hdp08:3306/sqoopdb", "sqoopdb.person", prop)

   //停止SparkContext

   sc.stop()

 }

}

1.用maven將程序打包


2.將Jar包提交到spark集群

/home/hadoop/apps/spark/bin/spark-submit \

--class net.togogo.sql.JdbcRDD \

--master spark://hdp08:7077 \

--jars /home/hadoop/mysql-connector-java-5.1.45.jar \

--driver-class-path /home/hadoop/mysql-connector-java-5.1.45.jar \

/home/hadoop/schema.jar

?

?

?

?

?

?

?

?

?

?

?

?

???

上一篇: {大數(shù)據(jù)}Spark Streaming

下一篇: {大數(shù)據(jù)}RDD分布式數(shù)據(jù)集

十九年老品牌
微信咨詢:gz_togogo 咨詢電話:18127429208 咨詢網(wǎng)站客服:在線客服

相關(guān)課程推薦

在線咨詢 ×

您好,請問有什么可以幫您?我們將竭誠提供最優(yōu)質(zhì)服務(wù)!