建議參考SparkSQL官方文檔:http://spark.apache.org/docs/latest/sql-programming-guide.htmlhtml
1、前述java
一、SparkSQL介紹mysql
Hive是Shark的前身,Shark是SparkSQL的前身,SparkSQL產生的根本緣由是其徹底脫離了Hive的限制。sql
二、Spark on Hive和Hive on Spark數據庫
Spark on Hive: Hive只做爲儲存角色,Spark負責sql解析優化,執行。apache
Hive on Spark:Hive即做爲存儲又負責sql的解析優化,Spark負責執行。編程
2、基礎概念json
一、DataFrame架構
DataFrame也是一個分佈式數據容器。與RDD相似,然而DataFrame更像傳統數據庫的二維表格,除了數據之外,還掌握數據的結構信息,即schema。同時,與Hive相似,DataFrame也支持嵌套數據類型(struct、array和map)。從API易用性的角度上 看, DataFrame API提供的是一套高層的關係操做,比函數式的RDD API要更加友好,門檻更低。分佈式
DataFrame的底層封裝的是RDD,只不過RDD的泛型是Row類型。
二、SparkSQL的數據源
SparkSQL的數據源能夠是JSON類型的字符串,JDBC,Parquent,Hive,HDFS等。
三、SparkSQL底層架構
首先拿到sql後解析一批未被解決的邏輯計劃,再通過分析獲得分析後的邏輯計劃,再通過一批優化規則轉換成一批最佳優化的邏輯計劃,再通過SparkPlanner的策略轉化成一批物理計劃,
隨後通過消費模型轉換成一個個的Spark任務執行。
四、謂詞下推(predicate Pushdown)
3、建立DataFrame的幾種方式
一、讀取json格式的文件建立DataFrame
SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("jsonfile"); SparkContext sc = new SparkContext(conf); //建立sqlContext SQLContext sqlContext = new SQLContext(sc);//SprakSQL中是SQLContext對象 /** * DataFrame的底層是一個一個的RDD RDD的泛型是Row類型。 * 如下兩種方式均可以讀取json格式的文件 */ DataFrame df = sqlContext.read().format("json").load("sparksql/json"); // DataFrame df2 = sqlContext.read().json("sparksql/json.txt"); // df2.show(); /** * DataFrame轉換成RDD */ RDD<Row> rdd = df.rdd(); /** * 顯示 DataFrame中的內容,默認顯示前20行。若是現實多行要指定多少行show(行數) * 注意:當有多個列時,顯示的列前後順序是按列的ascii碼前後顯示。 */ // df.show(); /** * 樹形的形式顯示schema信息 */ df.printSchema(); /** * dataFram自帶的API 操做DataFrame(很麻煩) */ //select name from table // df.select("name").show(); //select name age+10 as addage from table df.select(df.col("name"),df.col("age").plus(10).alias("addage")).show(); //select name ,age from table where age>19 df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show(); //select count(*) from table group by age df.groupBy(df.col("age")).count().show(); /** * 將DataFrame註冊成臨時的一張表,這張表臨時註冊到內存中,是邏輯上的表,不會霧化到磁盤 */ df.registerTempTable("jtable"); DataFrame sql = sqlContext.sql("select age,count(1) from jtable group by age"); DataFrame sql2 = sqlContext.sql("select * from jtable"); sc.stop();
二、經過json格式的RDD建立DataFrame
java代碼:
SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("jsonRDD"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList( "{\"name\":\"zhangsan\",\"age\":\"18\"}", "{\"name\":\"lisi\",\"age\":\"19\"}", "{\"name\":\"wangwu\",\"age\":\"20\"}" )); JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList( "{\"name\":\"zhangsan\",\"score\":\"100\"}", "{\"name\":\"lisi\",\"score\":\"200\"}", "{\"name\":\"wangwu\",\"score\":\"300\"}" )); DataFrame namedf = sqlContext.read().json(nameRDD); DataFrame scoredf = sqlContext.read().json(scoreRDD); namedf.registerTempTable("name"); scoredf.registerTempTable("score"); DataFrame result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name"); result.show(); sc.stop();
三、非json格式的RDD建立DataFrame(重要)
1) 經過反射的方式將非json格式的RDD轉換成DataFrame(不建議使用)
1.反序列化時serializable 版本號不一致時會致使不能反序列化。
2.子類中實現了serializable接口,父類中沒有實現,父類中的變量不能被序列化,序列化後父類中的變量會獲得null。
注意:父類實現serializable接口,子類沒有實現serializable接口時,子類能夠正常序列化
3.被關鍵字transient修飾的變量不能被序列化。
4.靜態變量不能被序列化,屬於類,不屬於方法和對象,因此不能被序列化。
另外:一個文件屢次writeObject時,若是有相同的對象已經寫入文件,那麼下次再寫入時,只保存第二次寫入的引用,讀取時,都是第一次保存的對象。
java代碼:
/** * 注意: * 1.自定義類必須是可序列化的 * 2.自定義類訪問級別必須是Public * 3.RDD轉成DataFrame會把自定義類中字段的名稱按assci碼排序 */ SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("RDD"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lineRDD = sc.textFile("sparksql/person.txt"); JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() { /** * */ private static final long serialVersionUID = 1L; @Override public Person call(String s) throws Exception { Person p = new Person(); p.setId(s.split(",")[0]); p.setName(s.split(",")[1]); p.setAge(Integer.valueOf(s.split(",")[2])); return p; } }); /** * 傳入進去Person.class的時候,sqlContext是經過反射的方式建立DataFrame * 在底層經過反射的方式得到Person的全部field,結合RDD自己,就生成了DataFrame */ DataFrame df = sqlContext.createDataFrame(personRDD, Person.class); df.show(); df.registerTempTable("person"); sqlContext.sql("select name from person where id = 2").show(); /** * 將DataFrame轉成JavaRDD * 注意: * 1.可使用row.getInt(0),row.getString(1)...經過下標獲取返回Row類型的數據,可是要注意列順序問題---不經常使用 * 2.可使用row.getAs("列名")來獲取對應的列值。 * */ JavaRDD<Row> javaRDD = df.javaRDD(); JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() { /** * */ private static final long serialVersionUID = 1L; @Override public Person call(Row row) throws Exception { Person p = new Person(); //p.setId(row.getString(1)); //p.setName(row.getString(2)); //p.setAge(row.getInt(0)); p.setId((String)row.getAs("id")); p.setName((String)row.getAs("name")); p.setAge((Integer)row.getAs("age")); return p; } }); map.foreach(new VoidFunction<Person>() { /** * */ private static final long serialVersionUID = 1L; @Override public void call(Person t) throws Exception { System.out.println(t); } }); sc.stop();
2) 動態建立Schema將非json格式的RDD轉換成DataFrame(建議使用)
SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("rddStruct"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt"); /** * 轉換成Row類型的RDD */ JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() { /** * */ private static final long serialVersionUID = 1L; @Override public Row call(String s) throws Exception { return RowFactory.create(//這裏字段順序必定要和下邊 StructField對應起來 String.valueOf(s.split(",")[0]), String.valueOf(s.split(",")[1]), Integer.valueOf(s.split(",")[2]) ); } }); /** * 動態構建DataFrame中的元數據,通常來講這裏的字段能夠來源自字符串,也能夠來源於外部數據庫 */ List<StructField> asList =Arrays.asList(//這裏字段順序必定要和上邊對應起來 DataTypes.createStructField("id", DataTypes.StringType, true), DataTypes.createStructField("name", DataTypes.StringType, true), DataTypes.createStructField("age", DataTypes.IntegerType, true) ); StructType schema = DataTypes.createStructType(asList); DataFrame df = sqlContext.createDataFrame(rowRDD, schema); df.show(); JavaRDD<Row> javaRDD = df.javaRDD(); javaRDD.foreach(new VoidFunction<Row>() { /** * */ private static final long serialVersionUID = 1L; @Override public void call(Row row) throws Exception {//Row類型的RDD System.out.println(row.getString(0)); } }) sc.stop();
四、讀取parquet文件建立DataFrame
注意:
df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet"); df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");
Overwrite:覆蓋
Append:追加
ErrorIfExists:若是存在就報錯
Ignore:若是存在就忽略
java代碼:
SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("parquet"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> jsonRDD = sc.textFile("sparksql/json"); DataFrame df = sqlContext.read().json(jsonRDD); /** * 將DataFrame保存成parquet文件,SaveMode指定存儲文件時的保存模式 * 保存成parquet文件有如下兩種方式: */ df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet"); df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet"); df.show(); /** * 加載parquet文件成DataFrame * 加載parquet文件有如下兩種方式: */ DataFrame load = sqlContext.read().format("parquet").load("./sparksql/parquet"); load = sqlContext.read().parquet("./sparksql/parquet"); load.show(); sc.stop()
五、讀取JDBC中的數據建立DataFrame(MySql爲例)
兩種方式建立DataFrame
java代碼:
SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("mysql"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); /** * 第一種方式讀取MySql數據庫表,加載爲DataFrame */ Map<String, String> options = new HashMap<String,String>(); options.put("url", "jdbc:mysql://192.168.179.4:3306/spark"); options.put("driver", "com.mysql.jdbc.Driver"); options.put("user", "root"); options.put("password", "123456"); options.put("dbtable", "person"); DataFrame person = sqlContext.read().format("jdbc").options(options).load(); person.show(); person.registerTempTable("person"); /** * 第二種方式讀取MySql數據表加載爲DataFrame */ DataFrameReader reader = sqlContext.read().format("jdbc"); reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark"); reader.option("driver", "com.mysql.jdbc.Driver"); reader.option("user", "root"); reader.option("password", "123456"); reader.option("dbtable", "score"); DataFrame score = reader.load(); score.show(); score.registerTempTable("score"); DataFrame result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name"); result.show(); /** * 將DataFrame結果保存到Mysql中 */ Properties properties = new Properties(); properties.setProperty("user", "root"); properties.setProperty("password", "123456"); result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties); sc.stop();
http://www.javashuo.com/article/p-obnhjwqq-gy.html
常見的action操做
一、count
count:對dataset中的記錄數進行統計個數的操做
二、first
first:獲取數據集中的第一條數據
三、reduce
reduce:對數據集中的全部數據進行歸約的操做,多條變成一條
四、show
show,默認將dataset數據打印前20條
五、take
take,從數據集中獲取指定條數
六、collect
collect:將分佈式存儲在集羣上的分佈式數據集(好比dataset),中的全部數據都獲取到driver端來
七、foreach
foreach:遍歷數據集中的每一條數據,對數據進行操做,這個跟collect不一樣,collect是將數據獲取到driver端進行操做
foreach是將計算操做推到集羣上去分佈式執行
foreach(println(_))這種,真正在集羣中執行的時候,是沒用的,由於輸出的結果是在分佈式的集羣中的,咱們是看不到的
-----------------------------------------------------------------------------------
//===========================================1 spark SQL=================== //數據導入方式 Dataset<Row> df = spark.read().json("..\\sparkTestData\\people.json"); //查看錶 df.show(); //查看錶結構 df.printSchema(); //查看某一列 相似於MySQL: select name from people df.select("name").show(); //查看多列並做計算 相似於MySQL: select name ,age+1 from people df.select(col("name"), col("age").plus(1)).show(); //設置過濾條件 相似於MySQL:select * from people where age>21 df.filter(col("age").gt(21)).show(); //作聚合操做 相似於MySQL:select age,count(*) from people group by age df.groupBy("age").count().show(); //上述多個條件進行組合 select ta.age,count(*) from (select name,age+1 as "age" from people) as ta where ta.age>21 group by ta.age df.select(col("name"), col("age").plus(1).alias("age")).filter(col("age").gt(21)).groupBy("age").count().show(); //直接使用spark SQL進行查詢 //先註冊爲臨時表 df.createOrReplaceTempView("people"); Dataset<Row> sqlDF = spark.sql("SELECT * FROM people"); sqlDF.show();