feat:线下课：创建DataFrame、DataFrameDSL、DataFrameSql、SparkOnHive

pom.xml

@@ -55,6 +55,20 @@
             <version>${spark.version}</version>
         </dependency>
 
+        <!-- Spark Hive -->
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-hive_2.11</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
+        <!-- Hive jdbc-->
+        <dependency>
+            <groupId>org.apache.hive</groupId>
+            <artifactId>hive-jdbc</artifactId>
+            <version>2.1.0</version>
+        </dependency>
+
         <!-- Hadoop Client -->
         <dependency>
             <groupId>org.apache.hadoop</groupId>

src/main/resources/spark/sql/student.csv

@@ -0,0 +1,4 @@
+id,name,age
+1,zhangsan,18
+2,lisi,18
+3,wangwu,18

src/main/scala/org/example/sql/CreateDataFrame.scala

@@ -1,7 +1,7 @@
 package org.example.sql
 
 import org.apache.spark.sql._
-
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
 
 object CreateDataFrame {
   def main(args: Array[String]): Unit = {
@@ -9,14 +9,54 @@ object CreateDataFrame {
       .master("local[*]")
       .appName("DataFrameDemo")
       .getOrCreate()
+    //    val data = List(
+    //      (1, "zhangsan", 18),
+    //      (2, "lisi", 18),
+    //      (3, "wangwu", 18)
+    //    )
+    //    import spark.implicits._
+    //    val df = data.toDF()
+    //    df.show()
+    //    df.printSchema()
+
+    // 实体类创建
+    println("=========实体类创建============")
     val data = List(
-      (1, "zhangsan", 18),
+      Row(1, "zhangsan", 18),
-      (2, "lisi", 18),
+      Row(2, "lisi", 18),
-      (3, "wangwu", 18)
+      Row(3, "wangwu", 18)
     )
-    import spark.implicits._
+    val idField = StructField("id", IntegerType, false)
-    val df = data.toDF()
+    val nameField = StructField("name", StringType, true)
+    val ageField = StructField("age", IntegerType, false)
+    val schema = StructType(List(idField, nameField, ageField))
+    val rdd = spark.sparkContext.parallelize(data)
+    val df = spark.createDataFrame(rdd, schema)
     df.show()
-    df.printSchema()
+
+
+    // 从外部文件创建DataFrame
+    println("=========外部文件============")
+    val dfCreate = spark.read
+      .option("header", "true")
+      .option("inferSchema", "true")
+      .csv("src/main/resources/Spark/sql/student.csv")
+
+    dfCreate.show()
+    dfCreate.printSchema()
+
+    // 通过jdbc读取数据库的表
+    println("=========jdbc============")
+    val dfSql = spark.read
+      .format("jdbc")
+      .option("url", "jdbc:mysql://192.168.48.10:3306/gmall")
+      .option("dbtable", "user_info")
+      .option("user", "root")
+      .option("password", "root123456")
+      .load()
+
+    dfSql.show()
+
+    spark.close()
   }
 }

src/main/scala/org/example/sql/DataFrameDSL.scala

@@ -0,0 +1,41 @@
+package org.example.sql
+
+import org.apache.spark
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions._
+object DataFrameDSL {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder()
+      .master("local[*]")
+      .appName("DataFrameDemo")
+      .getOrCreate()
+
+    val dfSql = spark.read
+      .format("jdbc")
+      .option("url", "jdbc:mysql://192.168.48.10:3306/gmall")
+      .option("dbtable", "user_info")
+      .option("user", "root")
+      .option("password", "root123456")
+      .load()
+
+    println("============show============")
+    dfSql.show()
+
+    println("============select============")
+    dfSql.select("login_name","name","gender").show()
+
+    println("============selectCol============")
+    dfSql.select(
+      col("login_name"),
+      col("name"),
+      col("user_level")
+    ).show()
+
+    println("============selectCol+1============")
+    dfSql.select(
+      col("login_name"),
+      col("name"),
+      (col("user_level") + 1).as("new_user_level")
+    ).show()
+  }
+}

src/main/scala/org/example/sql/DataFrameSql.scala

@@ -0,0 +1,35 @@
+package org.example.sql
+
+import org.apache.spark.sql.SparkSession
+
+object DataFrameSql {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder()
+      .master("local[*]")
+      .appName("DataFrameDemo")
+      .getOrCreate()
+
+    val dfSql = spark.read
+      .format("jdbc")
+      .option("url", "jdbc:mysql://192.168.48.10:3306/gmall")
+      .option("dbtable", "user_info")
+      .option("user", "root")
+      .option("password", "root123456")
+      .load()
+
+    println("============show============")
+    dfSql.show()
+
+    println("============sql============")
+    dfSql.createTempView("user_info")
+    spark.sql(
+      """
+        |select
+        |login_name
+        |,name
+        |,user_level
+        |from user_info
+        |""".stripMargin).show()
+
+  }
+}

src/main/scala/org/example/sql/SparkOnHive.scala

@@ -0,0 +1,18 @@
+package org.example.sql
+
+import org.apache.spark.sql.SparkSession
+
+object SparkOnHive {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder()
+      .appName("Spark On Hive")
+      .master("local")
+      .enableHiveSupport()
+      .getOrCreate()
+
+//    spark.sql("show databases").show()
+    spark.sql("use gmall").show()
+//    spark.sql("show tables").show()
+    spark.sql("select gender,count(1) cn from ods_user_info group by gender").show
+  }
+}