feat: 复习spark sql

Author: collabH

study-spark/spark-sql/src/main/scala/com/spark/sql/SparkHive.scala

@@ -1,20 +1,21 @@
 package com.spark.sql
 
+import org.apache.spark.sql.{DataFrame, SparkSession}
 import org.apache.spark.{SparkConf, SparkContext}
 
 /**
-  * @fileName: SparkHive.java
-  * @description: SparkHive.java类说明
-  * @author: by echo huang
-  * @date: 2020-06-29 10:27
-  */
+ * @fileName: SparkHive.java
+ * @description: SparkHive.java类说明
+ * @author: by echo huang
+ * @date: 2020-06-29 10:27
+ */
 object SparkHive extends App {
   override def main(args: Array[String]): Unit = {
-    testYarn()
+    //    testYarn()
     //    val sparkBuilder = SparkSession.builder()
     //      .master("local[*]")
     //      .appName("hive")
-    //      .config("spark.driver.memory", "4g")
+    //      .config"spark.driver.memory", "4g")
     //      .config("spark.num.executors", "4")
     //      .config("spark.executor.memory", "2g")
     //      .config("spark.executor.cores", "4")
@@ -26,6 +27,41 @@ object SparkHive extends App {
     //    val frame: DataFrame = spark.sql("select * from forchange_prod.user_orders")
     //    frame.show(20)
     //    spark.close()
+    val spark: SparkSession = SparkSession.builder().master("local[*]")
+      .appName("hive")
+      .config("spark.shuffle.manager", "sort")
+      .config("hive.exec.dynamic.partition", "true")
+      .config("hive.exec.dynamic.partition.mode", "nonstrict")
+      .config("hive.exec.max.dynamic.partitions", 2048)
+      .config("spark.sql.files.maxPartitionBytes", 134217728)
+      .config("spark.sql.shuffle.partitions", 200)
+      .config("spark.sql.inMemoryColumnarStorage.compressed", value = true)
+      // 是否启用bypass机制，如果分区数小于该则直接使用hash用于shuffle，前提shuffle map端没有预聚合操作
+      .config("spark.shuffle.sort.bypassMergeThreshold", 300)
+      .config("spark.shuffle.compress", value = true)
+      .config("spark.shuffle.file.buffer", "512k")
+      .config("spark.shuffle.io.numConnectionsPerPeer", 5)
+      .config("spark.shuffle.spill.compress", value = true)
+      .config("spark.io.compression.codec", "snappy")
+      .config("spark.driver.memory", "1g")
+      .config("spark.num.executors", "3")
+      .config("spark.executor.memory", "2g")
+      .config("spark.executor.cores", "3")
+      .config("spark.default.parallelism", "10")
+      .config("spark.mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
+      .config("spark.sql.parquet.writeLegacyFormat", "true")
+      .enableHiveSupport()
+      .getOrCreate()
+
+    //    spark.sql("show databases").show()
+
+    spark.sql("use wh_dwd")
+    spark.sql("show tables").show()
+
+    val startLog: DataFrame = spark.table("dwd_start_log")
+
+    startLog.show()
+    spark.stop()
   }
 
 
@@ -47,7 +83,7 @@ object SparkHive extends App {
       .set("spark.driver.host", "192.168.6.35")
       // 设置jar包的路径,如果有其他的依赖包,可以在这里添加,逗号隔开
       .setJars(List(""
-    ))
+      ))
     conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
     val sc = new SparkContext(conf)
     val input = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 0))