SparkSQL01

工作当中几乎全用SparkSQL ，RDD用的很少(面试多)
SparkSQL误区

Spark SQL is Apache Spark’s module for working with structured data.
不要把SparkSQL认为就是处理SQl的或者认为就是写SQL
SparkSQL

误区：
    1）Spark SQL是处理结构化数据
        并不是仅仅能够处理SQL
        SQL仅仅是Spark SQL这个模块的一小部分应用
        API/ExtDS
    2）Uniform Data Access  外部数据源(*****)
        Spark SQL是能够处理多种不同的数据源的数据
            text、json、parquet、orc、hive、jdbc    数据的格式 
            HDFS/S3(a/n)/OSS/COS                数据的存储系统
        不同的数据格式压缩的不压缩的 sparksql都是兼容的 
        你访问不同的数据源SparkSQl都是用统一的访问方式  这就是外部数据源

SparkSQL能面试的东西 就是两个 ：
	DataFrame 、 外部数据源、catelist 

2.能集成Hive
你的数仓以前是基于Hive来做的 都是Hive的脚本 
 现在 如果想使用SparkSQL访问Hive的数据 SparkSQL能连接到MetaStore才可以
 (把Hive-site.xml  拷贝到Sparkconf目录下就可以了)
 因为MetaStore 是 on Hadoop的核心所在 

所以你要把Hive迁移到Spark上来 成本是很低的

在这里插入图片描述

3.Standard Connectivity
Hive能通过HiveServer2提供一个服务 大家去查，那么 spark里面有个thriftServer 
他们底层都是用thrift协议的

在这里插入图片描述

误区3：
MR==>Hive==>  Hive底层当时是MR 慢 所以出来Spark 
     Spark==> AMPLab Shark(为了将Hive SQL跑在Spark上)  1.x  配套一个打了补丁的Hive
        Spark1.0  Shark不维护
            ==> Spark SQL 是在Spark里面的
            ==> Hive on Spark 是在Hive里面的      是Hive的引擎是Spark

误区3）
    Hive on Spark不是Spark SQL
        Hive刚开始时底层执行引擎只有一个：MR
        后期：Tez Spark
        set hive.execution.engine=spark;    就可以 Hive on Spark

    SparkSQL on Hive  X

Hive On Spark

Time taken: 6.86 seconds, Fetched: 2 row(s)
hive (default)> set hive.execution.engine;
hive.execution.engine=mr
hive (default)> set hive.execution.engine=spark;
hive (default)> set hive.execution.engine;
hive.execution.engine=spark
hive (default)> show databases;
OK
database_name
default
homework
Time taken: 0.008 seconds, Fetched: 2 row(s)
hive (default)>

这个东西了解即可 Hive On Spark 真正生产上用的很少的 
这个东西不是很成熟的

Datasets and DataFrames

出来的时间：

Spark SQL
    1.0     
    SchemaRDD  ==> Table     RDD(存数据) + schema = Table
    ==> DataFrame  1.2/3     由SchemaRDD  变为DataFrame 原因是 更加 OO
    ==> Dataset    1.6    由DataFrame  变为Dataset 因为 compile-time type safety

DataFrame
A Dataset is a distributed collection of data.
A DataFrame is a Dataset organized into named columns.
DataFrame = Dataset[Row]
In Scala and Java, a DataFrame is represented by a Dataset of Rows.

DataFrame ：
	1.named columns    就是一个表  包含 列的名字 + 列的类型 
	
Row ： 可以理解为  一行数据 没有scheme的 

SparkSession是Spark编程的入口点

在这里插入图片描述
Api：

SparkSession：
  /**
   * Executes a SQL query using Spark, returning the result as a `DataFrame`.
   * The dialect that is used for SQL parsing can be configured with 'spark.sql.dialect'.
   *
   * @since 2.0.0
   */
  def sql(sqlText: String): DataFrame = {
    Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText))
  }

注意:
	1. returning the result as a `DataFrame`


Dataset：
  /**
   * Displays the top 20 rows of Dataset in a tabular form.
   *
   * @param truncate Whether truncate long strings. If true, strings more than 20 characters will
   *                 be truncated and all cells will be aligned right
   *
   * @group action
   * @since 1.6.0
   */
  def show(truncate: Boolean): Unit = show(20, truncate)

scala> spark.sql("show tables").show

+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
| default|  student|      false|
+--------+---------+-----------+

scala> 

注意：
	启动spark-shell的时候  指定MySQL驱动  
	个人建议使用 --jars 指定MySQL驱动 
	不建议把MySQL驱动 直接丢在Spark jar路径里

查看Hive里元数据：

mysql> select * from DBS;
+-------+-----------------------+-------------------------------------------------------+----------+--------------+------------+
| DB_ID | DESC                  | DB_LOCATION_URI                                       | NAME     | OWNER_NAME   | OWNER_TYPE |
+-------+-----------------------+-------------------------------------------------------+----------+--------------+------------+
|     1 | Default Hive database | hdfs://hadoop101:8020/user/hive/warehouse             | default  | public       | ROLE       |
|     6 | NULL                  | hdfs://hadoop101:8020/user/hive/warehouse/homework.db | homework | double_happy | USER       |
+-------+-----------------------+-------------------------------------------------------+----------+--------------+------------+
2 rows in set (0.00 sec)

mysql> select * from TBLS;
+--------+-------------+-------+------------------+--------------+-----------+-------+-----------------------------------+----------------+--------------------+--------------------+
| TBL_ID | CREATE_TIME | DB_ID | LAST_ACCESS_TIME | OWNER        | RETENTION | SD_ID | TBL_NAME                          | TBL_TYPE       | VIEW_EXPANDED_TEXT | VIEW_ORIGINAL_TEXT |
+--------+-------------+-------+------------------+--------------+-----------+-------+-----------------------------------+----------------+--------------------+--------------------+
|      1 |  1568615059 |     1 |                0 | double_happy |         0 |     1 | student                           | MANAGED_TABLE  | NULL               | NULL               |
|      8 |  1568616039 |     6 |                0 | double_happy |         0 |     8 | ods_domain_traffic_info           | EXTERNAL_TABLE | NULL               | NULL               |
|      9 |  1568620410 |     6 |                0 | double_happy |         0 |     9 | ods_uid_pid_info                  | EXTERNAL_TABLE | NULL               | NULL               |
|     17 |  1568860945 |     6 |                0 | double_happy |         0 |    17 | jf_tmp                            | MANAGED_TABLE  | NULL               | NULL               |
|     21 |  1569056727 |     6 |                0 | double_happy |         0 |    21 | access_wide                       | EXTERNAL_TABLE | NULL               | NULL               |
|     26 |  1569209493 |     6 |                0 | double_happy |         0 |    31 | ods_uid_pid_info_compression_test | EXTERNAL_TABLE | NULL               | NULL               |
|     27 |  1569209946 |     6 |                0 | double_happy |         0 |    32 | ods_uid_pid_compression_info      | MANAGED_TABLE  | NULL               | NULL               |
|     31 |  1569224142 |     6 |                0 | double_happy |         0 |    36 | dwd_platform_stat_info            | MANAGED_TABLE  | NULL               | NULL               |
|     53 |  1570957119 |     6 |                0 | double_happy |         0 |    63 | ods_log_info                      | EXTERNAL_TABLE | NULL               | NULL               |
+--------+-------------+-------+------------------+--------------+-----------+-------+-----------------------------------+----------------+--------------------+--------------------+
9 rows in set (0.00 sec)

mysql> 


spark-shell查询Hive里的表：

scala> spark.sql("select * from homework.dwd_platform_stat_info").show
+--------+---+--------+--------+                                                
|platform|cnt|       d|     day|
+--------+---+--------+--------+
| Andriod|658|20190921|20190921|
| Symbain|683|20190921|20190921|
|   linux|639|20190921|20190921|
|     mac|652|20190921|20190921|
| windows|640|20190921|20190921|
+--------+---+--------+--------+

scala>

使用sparksql 在spark-shell交互 还得写 spark.sql
在spark里 有个 spark-sql  用法和 spark-shell 是一样的

编程

sparksql编程

1.SparkSession构建

object SparkSessionApp {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()


    spark.stop()
  }
}

当然 你spark一些参数如何传进去呢？
提供config传进去
eg ： 你要设置多少个分区呀 等

在这里插入图片描述
Data Sources

1.读文本数据

1.读文本数据

object SparkSessionApp {

  def text(spark: SparkSession) = {
    import spark.implicits._
    val df: DataFrame = spark.read.format("text").load("file:///C:/IdeaProjects/spark/data/data.txt")
    df.show()
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    text(spark)

    spark.stop()
  }

}

结果：
+---------------+
|          value|
+---------------+
|double_happy,25|
|      Kairis,25|
|        Kite,32|
+---------------+

1. 但是有一个问题 读取进来的数据   把所有内容
都放到 value这个列 里面去了 
该怎么办？

2. 上面那种写法读进来的是DF

def text(spark: SparkSession) = {
    val ds: Dataset[String] = spark.read.textFile("file:///C:/IdeaProjects/spark/data/data.txt")
        ds.show()
  }

读进来的是DS
结果是一样的：
+---------------+
|          value|
+---------------+
|double_happy,25|
|      Kairis,25|
|        Kite,32|
+---------------+


 /**
   * Loads text files and returns a [[Dataset]] of String. See the documentation on the
   * other overloaded `textFile()` method for more details.
   * @since 2.0.0
   */
  def textFile(path: String): Dataset[String] = {
    // This method ensures that calls that explicit need single argument works, see SPARK-16009
    textFile(Seq(path): _*)
  }

可以传入多个路径的    textFile(Seq(path): _*)

取出第一列输出出去注意df 和ds的区别

df：
在这里插入图片描述

1. df.map  里面是row     ds.map  里面是String
 2. ds 可以map 里面 x.split 
   df 就不可以 
 那我要取出第一列使用df 该这么办？
这就是 df 和 ds 编程的 最本质的区别   df = ds[Row]

所以 df 得使用  df.rdd.map  
而且他的返回值是 rdd

在这里插入图片描述

object SparkSessionApp2 {

  def text(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("text").load("file:///C:/IdeaProjects/spark/data/data.txt")

    val ds: Dataset[String] = spark.read.textFile("file:///C:/IdeaProjects/spark/data/data.txt")

    val result: RDD[(String, String)] = df.rdd.map(x => {
      val tmp: String = x.getString(0)
      val splits: Array[String] = tmp.split(",")
      (splits(0), splits(1))
    })
    result.foreach(println)

  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    text(spark)
    spark.stop()
  }

}

结果是：
(double_happy,25)
(Kairis,25)
(Kite,32)

这个结果不是我们想要的 ，我要的是 把结果写出去 
上面这种是 df的 那么 ds该怎么操作呢？

object SparkSessionApp2 {

  def text(spark: SparkSession) = {

    val ds: Dataset[String] = spark.read.textFile("file:///C:/IdeaProjects/spark/data/data.txt")

    import spark.implicits._
    val resultDS: Dataset[(String, String)] = ds.map(x => {
      val splits: Array[String] = x.split(",")
      (splits(0), splits(1))
    })

    resultDS.write.mode(SaveMode.Overwrite).format("text").save("file:///C:/IdeaProjects/spark/out-sparksql")

  }
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    text(spark)


    spark.stop()
  }
}

在这里插入图片描述

那么我们只输出一列 ：
object SparkSessionApp2 {

  def text(spark: SparkSession) = {

    val ds: Dataset[String] = spark.read.textFile("file:///C:/IdeaProjects/spark/data/data.txt")

    import spark.implicits._
    val resultDS = ds.map(x => {
      val splits: Array[String] = x.split(",")
      splits(0)
    })
    resultDS.write.mode(SaveMode.Overwrite).format("text").save("file:///C:/IdeaProjects/spark/out-sparksql")
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()
    text(spark)
    spark.stop()
  }
}
结果 ：ok

在这里插入图片描述

但是 有个问题的 文本格式是非常常用的格式  你只支持  一列输出 有个鬼用

这个问题该这么解决呢？
  这个问题很重要 前面的 不同类型日志输出  一定是多列的  
  下面讲到压缩  给你一个场景 
  andriod 的 bzip的  ios gzip 的   windos bz2  你该这么办？  这都是常见的需求

上面的问题之后再解决

那么这个输出的数据也是可以用压缩的

object SparkSessionApp2 {

  def text(spark: SparkSession) = {
    val ds: Dataset[String] = spark.read.textFile("file:///C:/IdeaProjects/spark/data/data.txt")

    import spark.implicits._
    val resultDS = ds.map(x => {
      val splits: Array[String] = x.split(",")
      (splits(0))
    })
   resultDS.write.option("compression","gzip").mode(SaveMode.Overwrite).format("text").save("file:///C:/IdeaProjects/spark/out-sparksql")
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    text(spark)
    spark.stop()
  }

}
结果：

在这里插入图片描述

注意：
在这里插入图片描述

也就是说这个压缩 codec 是有限制的 
问题：让是输出使用lzo 压缩该怎么办呢？

2.读json数据

在这里插入图片描述

object SparkSessionApp2 {

  def json(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("json").load("file:///C:/IdeaProjects/spark/data/data.log")

    df.printSchema()
    df.show()
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    json(spark)
    spark.stop()
  }
}

结果是：
root
 |-- _corrupt_record: string (nullable = true)
 |-- age: long (nullable = true)
 |-- branch: string (nullable = true)
 |-- camera_id: string (nullable = true)
 |-- camera_ip: string (nullable = true)
 |-- client_time: struct (nullable = true)
 |    |-- enter_time: long (nullable = true)
 |    |-- exit_time: long (nullable = true)
 |    |-- first_time: long (nullable = true)
 |    |-- last_time: long (nullable = true)
 |-- events: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- host_time: long (nullable = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- osd_time: long (nullable = true)
 |-- face_id: string (nullable = true)
 |-- gender: long (nullable = true)
 |-- is_new_user: boolean (nullable = true)
 |-- mall_id: string (nullable = true)
 |-- match_photo_index: long (nullable = true)
 |-- match_score: long (nullable = true)
 |-- package_index: long (nullable = true)
 |-- photos: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- frame_time: long (nullable = true)
 |    |    |-- quality: double (nullable = true)
 |    |    |-- url: string (nullable = true)
 |-- process_context: struct (nullable = true)
 |    |-- history_res: string (nullable = true)
 |    |-- temp_res: string (nullable = true)
 |-- process_end_time: long (nullable = true)
 |-- process_start_time: long (nullable = true)
 |-- product_id: string (nullable = true)
 |-- project_id: string (nullable = true)
 |-- race: long (nullable = true)
 |-- request_id: string (nullable = true)
 |-- request_time: long (nullable = true)
 |-- site_id: string (nullable = true)
 |-- status: long (nullable = true)
 |-- temp_id: string (nullable = true)
 |-- tracks: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- box: struct (nullable = true)
 |    |    |    |-- angle: long (nullable = true)
 |    |    |    |-- height: long (nullable = true)
 |    |    |    |-- left: long (nullable = true)
 |    |    |    |-- top: long (nullable = true)
 |    |    |    |-- width: long (nullable = true)
 |    |    |-- host_time: long (nullable = true)
 |    |    |-- index: long (nullable = true)
 |    |    |-- video_time: long (nullable = true)
 |-- user_id: string (nullable = true)

19/10/28 10:17:49 WARN Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
+---------------+---+--------------------+--------------------+-----------+--------------------+--------------------+--------------------+------+-----------+------------------+-----------------+-----------+-------------+--------------------+---------------+----------------+------------------+-----------+--------------------+----+--------------------+-------------+------------------+------+-------+--------------------+-------+
|_corrupt_record|age|              branch|           camera_id|  camera_ip|         client_time|              events|             face_id|gender|is_new_user|           mall_id|match_photo_index|match_score|package_index|              photos|process_context|process_end_time|process_start_time| product_id|          project_id|race|          request_id| request_time|           site_id|status|temp_id|              tracks|user_id|
+---------------+---+--------------------+--------------------+-----------+--------------------+--------------------+--------------------+------+-----------+------------------+-----------------+-----------+-------------+--------------------+---------------+----------------+------------------+-----------+--------------------+----+--------------------+-------------+------------------+------+-------+--------------------+-------+
|           null|  0|low_quality_faceI...|afu-hanghai-yxhqg...|172.16.10.2|[1555054289000, 1...|[[1555073266644, ...|PROJAFU_shanghai_...|     0|       true|AFU_shanghai_yxhqg|                0|          0|            9|[[1555054284000, ...|   [null, null]|   1555073288125|     1555073288099|trafficfull|AFU_shanghai_yxhq...|   0|f0cbcac5-60aa-498...|1555073288097|AFU_shanghai_yxhqg|    -1|       |[[[-21, 62, 884, ...|       |
|           null| 23|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055539...|[[1555073289722, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                0|          0|           -1|[[1555055540000, ...|   [null, null]|   1555073292530|     1555073292487|trafficfull|AFU_beijing_xhm_t...|   0|b264f039-431e-4c9...|1555073292487|   AFU_beijing_xhm|     4|       |[[[0, 74, 1656, 1...|       |
|           null| 21|  low_quality_faceId|afu-hanghai-yxhqg...|172.16.10.2|[0, 0, 1555054302...|[[1555073285646, ...|PROJAFU_shanghai_...|     0|       true|AFU_shanghai_yxhqg|                2|          0|            3|[[1555054311000, ...|   [null, null]|   1555073297234|     1555073297137|trafficfull|AFU_shanghai_yxhq...|   1|9e9b0963-96d9-44b...|1555073297136|AFU_shanghai_yxhqg|    -1|       |[[[12, 88, 674, 4...|       |
|           null| 22|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055539...|[[1555073289893, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                0|          0|            1|[[1555055543000, ...|   [null, null]|   1555073298078|     1555073298034|trafficfull|AFU_beijing_xhm_t...|   0|8703d95e-8ca2-4a8...|1555073298034|   AFU_beijing_xhm|    -1|       |[[[7, 72, 171, 56...|       |
|           null| 21|  low_quality_faceId|afu-hanghai-yxhqg...|172.16.10.2|[0, 0, 1555054302...|[[1555073285646, ...|PROJAFU_shanghai_...|     0|       true|AFU_shanghai_yxhqg|                2|          0|            5|[[1555054311000, ...|   [null, null]|   1555073300572|     1555073300471|trafficfull|AFU_shanghai_yxhq...|   1|163a8256-d832-427...|1555073300471|AFU_shanghai_yxhqg|    -1|       |[[[19, 96, 625, 3...|       |
|           null|  0|        empty_photos|afu-beijing-cytj-...|172.16.10.2|[0, 0, 1555068460...|[[1555073300056, ...|PROJAFU_beijing_c...|     0|      false|  AFU_beijing_cytj|                0|          0|           -1|                null|           [, ]|   1555073300572|     1555073300572|trafficfull|AFU_beijing_cytj_...|   0|5499d569-4067-42d...|1555073300494|  AFU_beijing_cytj|     4|       |[[[26, 55, 1341, ...|       |
|           null| 25|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[1555055520000, 1...|[[1555073271554, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                0|          0|           14|[[1555055520000, ...|   [null, null]|   1555073301353|     1555073301300|trafficfull|AFU_beijing_xhm_t...|   1|80f35edb-e3c9-48f...|1555073301299|   AFU_beijing_xhm|    -1|       |[[[25, 110, 1269,...|       |
|           null| 29|  low_quality_faceId|afu-beijing-cytj-...|172.16.10.2|[0, 0, 1555068461...|[[1555073300728, ...|PROJAFU_beijing_c...|     0|       true|  AFU_beijing_cytj|                0|          0|           -1|[[1555068461000, ...|   [null, null]|   1555073302108|     1555073302059|trafficfull|AFU_beijing_cytj_...|   1|ddbacad2-cd35-4a1...|1555073302058|  AFU_beijing_cytj|     4|       |[[[1, 108, 825, 1...|       |
|           null| 21|  low_quality_faceId|afu-hanghai-yxhqg...|172.16.10.2|[0, 0, 1555054302...|[[1555073285646, ...|PROJAFU_shanghai_...|     0|       true|AFU_shanghai_yxhqg|                2|          0|            6|[[1555054311000, ...|   [null, null]|   1555073302221|     1555073302126|trafficfull|AFU_shanghai_yxhq...|   1|3b0353b3-d5ec-492...|1555073302125|AFU_shanghai_yxhqg|    -1|       |[[[4, 85, 647, 39...|       |
|           null| 25|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055554...|[[1555073303646, ...|PROJAFU_beijing_x...|     1|       true|   AFU_beijing_xhm|                0|          0|           -1|[[1555055554000, ...|   [null, null]|   1555073305191|     1555073305148|trafficfull|AFU_beijing_xhm_t...|   0|3bd7e125-ac80-4ff...|1555073305148|   AFU_beijing_xhm|     4|       |[[[11, 63, 925, 1...|       |
|           null| 25|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[1555055520000, 1...|[[1555073271554, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                0|          0|           17|[[1555055520000, ...|   [null, null]|   1555073306338|     1555073306297|trafficfull|AFU_beijing_xhm_t...|   1|26383dcd-47a4-410...|1555073306297|   AFU_beijing_xhm|    -1|       |[[[11, 101, 1254,...|       |
|           null| 26|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055563...|[[1555073312893, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                1|          0|           -1|[[1555055564000, ...|   [null, null]|   1555073314733|     1555073314663|trafficfull|AFU_beijing_xhm_t...|   0|e8a517a4-bf72-46f...|1555073314663|   AFU_beijing_xhm|     4|       |[[[0, 102, 554, 2...|       |
|           null|  0|        empty_photos|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055566...|[[1555073315646, ...|PROJAFU_beijing_x...|     0|      false|   AFU_beijing_xhm|                0|          0|           -1|                null|           [, ]|   1555073315801|     1555073315801|trafficfull|AFU_beijing_xhm_t...|   0|1256cd65-3100-448...|1555073315797|   AFU_beijing_xhm|     4|       |[[[-8, 79, 1638, ...|       |
|           null| 25|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[1555055520000, 1...|[[1555073271554, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                0|          0|           23|[[1555055520000, ...|   [null, null]|   1555073316495|     1555073316453|trafficfull|AFU_beijing_xhm_t...|   1|292ea3d7-cc3b-452...|1555073316453|   AFU_beijing_xhm|    -1|       |[[[27, 99, 1243, ...|       |
|           null|  0|        empty_photos|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055567...|[[1555073316558, ...|PROJAFU_beijing_x...|     0|      false|   AFU_beijing_xhm|                0|          0|           -1|                null|           [, ]|   1555073316856|     1555073316856|trafficfull|AFU_beijing_xhm_t...|   0|caf82eb1-8f49-485...|1555073316856|   AFU_beijing_xhm|     4|       |[[[0, 68, 1695, 3...|       |
|           null|  0|        empty_photos|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055567...|[[1555073316313, ...|PROJAFU_beijing_x...|     0|      false|   AFU_beijing_xhm|                0|          0|           -1|                null|           [, ]|   1555073317218|     1555073317218|trafficfull|AFU_beijing_xhm_t...|   0|868a5c14-903e-461...|1555073317129|   AFU_beijing_xhm|     4|       |[[[9, 129, 993, 3...|       |
|           null| 31|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[1555055564000, 1...|[[1555073302556, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                2|          0|            5|[[1555055558000, ...|   [null, null]|   1555073317592|     1555073317503|trafficfull|AFU_beijing_xhm_t...|   0|bb3b4831-f1db-4f9...|1555073317503|   AFU_beijing_xhm|    -1|       |[[[6, 91, 172, 56...|       |
|           null|  0|        empty_photos|afu-beijing-xhm-c...|172.16.10.2|[0, 0, 1555055569...|[[1555073318311, ...|PROJAFU_beijing_x...|     0|      false|   AFU_beijing_xhm|                0|          0|           -1|                null|           [, ]|   1555073318529|     1555073318529|trafficfull|AFU_beijing_xhm_t...|   0|312c43a4-a247-464...|1555073318529|   AFU_beijing_xhm|     4|       |[[[2, 73, 1024, 1...|       |
|           null| 31|  low_quality_faceId|afu-beijing-xhm-c...|172.16.10.2|[1555055564000, 1...|[[1555073302556, ...|PROJAFU_beijing_x...|     0|       true|   AFU_beijing_xhm|                2|          0|            7|[[1555055558000, ...|   [null, null]|   1555073320823|     1555073320721|trafficfull|AFU_beijing_xhm_t...|   0|da438ef2-daf6-472...|1555073320721|   AFU_beijing_xhm|    -1|       |[[[18, 105, 186, ...|       |
|           null| 21|  low_quality_faceId|afu-hanghai-yxhqg...|172.16.10.2|[0, 0, 1555054302...|[[1555073285646, ...|PROJAFU_shanghai_...|     0|       true|AFU_shanghai_yxhqg|                2|          0|          -18|[[1555054311000, ...|   [null, null]|   1555073321796|     1555073321700|trafficfull|AFU_shanghai_yxhq...|   1|d16a278f-3ae9-44a...|1555073321700|AFU_shanghai_yxhqg|     4|       |[[[10, 58, 598, 3...|       |
+---------------+---+--------------------+--------------------+-----------+--------------------+--------------------+--------------------+------+-----------+------------------+-----------------+-----------+-------------+--------------------+---------------+----------------+------------------+-----------+--------------------+----+--------------------+-------------+------------------+------+-------+--------------------+-------+
only showing top 20 rows

def json(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("json").load("file:///C:/IdeaProjects/spark/data/data.log")


    df.select("age","branch","mall_id").filter("is_new_user = true ").show(10)
    
    df.select("age","branch","mall_id").where("is_new_user = true").show(10)

  }

结果是一样的哈 

/**
   * Filters rows using the given SQL expression.
   * {{{
   *   peopleDs.where("age > 15")
   * }}}
   *
   * @group typedrel
   * @since 1.6.0
   */
  def where(conditionExpr: String): Dataset[T] = {
    filter(Column(sparkSession.sessionState.sqlParser.parseExpression(conditionExpr)))
  }

where 底层调用的是 filter


结果：
+---+--------------------+------------------+
|age|              branch|           mall_id|
+---+--------------------+------------------+
|  0|low_quality_faceI...|AFU_shanghai_yxhqg|
| 23|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 22|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 29|  low_quality_faceId|  AFU_beijing_cytj|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
+---+--------------------+------------------+
only showing top 10 rows

+---+--------------------+------------------+
|age|              branch|           mall_id|
+---+--------------------+------------------+
|  0|low_quality_faceI...|AFU_shanghai_yxhqg|
| 23|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 22|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 29|  low_quality_faceId|  AFU_beijing_cytj|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
+---+--------------------+------------------+
only showing top 10 rows

filter 和where 里面 有好多中写法 ：
个人喜欢使用 '列名 +判断条件

在这里插入图片描述

但是报错：  加一个隐式转换

在这里插入图片描述

 def json(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("json").load("file:///C:/IdeaProjects/spark/data/data.log")
    import spark.implicits._
    df.select("age","branch","mall_id").filter('is_new_user === "true").show(10)
  }
结果是：
+---+--------------------+------------------+
|age|              branch|           mall_id|
+---+--------------------+------------------+
|  0|low_quality_faceI...|AFU_shanghai_yxhqg|
| 23|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 22|  low_quality_faceId|   AFU_beijing_xhm|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 29|  low_quality_faceId|  AFU_beijing_cytj|
| 21|  low_quality_faceId|AFU_shanghai_yxhqg|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
| 25|  low_quality_faceId|   AFU_beijing_xhm|
+---+--------------------+------------------+
only showing top 10 rows

在这里插入图片描述

写法很多 ：
 def json(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("json").load("file:///C:/IdeaProjects/spark/data/data.log")
    df.select("age","branch","mall_id").filter(df.col("is_new_user") === "true").show(10)
  }
结果是一样的


我个人是喜欢 
import spark.implicits._
select("age","branch","mall_id")  +  .filter('is_new_user === "true")    

这样写代码量少一些

object SparkSessionApp2 {

  def json(spark: SparkSession) = {
    val df: DataFrame = spark.read.format("json").load("file:///C:/IdeaProjects/spark/data/data.log")

    import spark.implicits._
    val resultDF: Dataset[Row] = df.select("age","branch","mall_id").filter('is_new_user === "true")

    resultDF.write
      .mode(SaveMode.Overwrite)
      .format("json")
      .save("file:///C:/IdeaProjects/spark/out-sparksql-json")

  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    json(spark)
    spark.stop()
  }
}

结果：

在这里插入图片描述

解析json 嵌套 + Sturct类型的 你会么？  给个思路 就是  exploded +打点

3.读csv数据
csv文件打开是execel能看见的
在这里插入图片描述

object SparkSessionApp2 {

 
  def csv(spark: SparkSession) = {

    val df: DataFrame = spark.read.format("csv").load("file:///C:/IdeaProjects/spark/data/data.csv")

    df.printSchema()
    df.show(10)

  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

     csv(spark)
    spark.stop()
  }

}

结果是：
root
 |-- _c0: string (nullable = true)

+--------------------+
|                 _c0|
+--------------------+
|pid	pid_type	stor...|
|2637034	GLOBAL	30...|
|127599	GLOBAL	303...|
|2626026	GLOBAL	30...|
|2643291	GLOBAL	30...|
|182310	GLOBAL	303...|
|182310	GLOBAL	303...|
|856248	GLOBAL	303...|
|29052	GLOBAL	3039...|
|29052	GLOBAL	3039...|
+--------------------+
only showing top 10 rows


所以这种处理结果不是我们想要的 
所以处理 csv 文件的时候 需要一些 option 需要我们添加的

object SparkSessionApp2 {


  def csv(spark: SparkSession) = {

    val df = spark.read.option("header","true")
      .format("csv").load("file:///C:/IdeaProjects/spark/data/data.csv")

    df.printSchema()
    println("......")
    df.show(10)

  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()
     csv(spark)
    spark.stop()
  }
}

结果是：
root
 |-- pid	pid_type	store_id	store_name	floor	start_time	end_time	event_type	label_version	channel: string (nullable = true)

......
+-------------------------------------------------------------------------------------------+
|pid	pid_type	store_id	store_name	floor	start_time	end_time	event_type	label_version	channel|
+-------------------------------------------------------------------------------------------+
|                                                                       2637034	GLOBAL	30...|
|                                                                       127599	GLOBAL	303...|
|                                                                       2626026	GLOBAL	30...|
|                                                                       2643291	GLOBAL	30...|
|                                                                       182310	GLOBAL	303...|
|                                                                       182310	GLOBAL	303...|
|                                                                       856248	GLOBAL	303...|
|                                                                       29052	GLOBAL	3039...|
|                                                                       29052	GLOBAL	3039...|
|                                                                       29052	GLOBAL	3039...|
+-------------------------------------------------------------------------------------------+
only showing top 10 rows

表 头出来了  但是不是我们想要的
这个头 就一列  没有分开  所以 还得加option  把头拆开

object SparkSessionApp2 {

  def csv(spark: SparkSession) = {

    val df = spark.read.option("header","true")
      .option("sep","\t")
      .format("csv").load("file:///C:/IdeaProjects/spark/data/data.csv")

    df.printSchema()
    println("......")
    df.show(10)
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()
     csv(spark)
    spark.stop()
  }
}

结果是：
root
 |-- pid: string (nullable = true)
 |-- pid_type: string (nullable = true)
 |-- store_id: string (nullable = true)
 |-- store_name: string (nullable = true)
 |-- floor: string (nullable = true)
 |-- start_time: string (nullable = true)
 |-- end_time: string (nullable = true)
 |-- event_type: string (nullable = true)
 |-- label_version: string (nullable = true)
 |-- channel: string (nullable = true)

......
+-------+--------+--------+--------------+-----+----------+--------+----------+--------------------+-------+
|    pid|pid_type|store_id|    store_name|floor|start_time|end_time|event_type|       label_version|channel|
+-------+--------+--------+--------------+-----+----------+--------+----------+--------------------+-------+
|2637034|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  17:38:44|17:39:32|         0|3b47b5f2c1d95c2fb...|ch11001|
| 127599|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  20:09:26|20:18:03|         0|3b47b5f2c1d95c2fb...|ch11001|
|2626026|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  11:38:21|11:38:50|         0|3b47b5f2c1d95c2fb...|ch11001|
|2643291|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  21:07:31|21:09:01|         0|3b47b5f2c1d95c2fb...|ch11001|
| 182310|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  10:41:34|10:41:55|         0|3b47b5f2c1d95c2fb...|ch11001|
| 182310|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  10:42:02|10:57:19|         0|3b47b5f2c1d95c2fb...|ch11001|
| 856248|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  14:49:23|14:56:18|         0|3b47b5f2c1d95c2fb...|ch11001|
|  29052|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  13:12:00|13:13:57|         0|3b47b5f2c1d95c2fb...|ch11001|
|  29052|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  14:14:28|14:14:55|         0|3b47b5f2c1d95c2fb...|ch11001|
|  29052|  GLOBAL|   3039A|Onitsuka Tiger|   3F|  14:30:38|14:30:52|         0|3b47b5f2c1d95c2fb...|ch11001|
+-------+--------+--------+--------------+-----+----------+--------+----------+--------------------+-------+
only showing top 10 rows

object SparkSessionApp2 {

  def csv(spark: SparkSession) = {

    val df = spark.read.option("header","true")
      .option("sep","\t")
      .format("csv").load("file:///C:/IdeaProjects/spark/data/data.csv")

    df.printSchema()
    println("......")
    df.show(10)

    import spark.implicits._
    df.select("pid","store_name").filter($"store_id" === "3039A")
      .write.format("csv").mode("overwrite")
      .save("file:///C:/IdeaProjects/spark/out-sparksql-csv")
  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

     csv(spark)
    spark.stop()
  }
}

结果是：

在这里插入图片描述

这些 option 参数 我是怎么知道的 ？去源码里找 
CSVOptions 类下面

4.读jdbc数据
MySQL中的数据是这样的
在这里插入图片描述

object SparkSessionApp2 {

  def jdbc(spark: SparkSession) = {
    import spark.implicits._

    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://hadoop101:3306")
      .option("dbtable", "hive_dwd.topn")
      .option("user", "root")
      .option("password", "wsx123$%^")
      .load()

        jdbcDF.show()

  }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    jdbc(spark)
    spark.stop()
  }
}

结果是:
+---------------+-----+---+
|         domain|  url|cnt|
+---------------+-----+---+
|  www.baidu.com| url5|  5|
|  www.baidu.com| url2|  2|
|  www.baidu.com| url4|  4|
|  www.baidu.com| url1|  1|
|  www.baidu.com| url3|  3|
|www.twitter.com| url6|  1|
|www.twitter.com|url10| 11|
|www.twitter.com| url9|  6|
| www.google.com| url2|  2|
| www.google.com| url6|  7|
| www.google.com| url1|  1|
| www.google.com| url8|  7|
+---------------+-----+---+

JDBC To Other Databases
官网有好多写法

object SparkSessionApp2 {
  def jdbc(spark: SparkSession) = {
    import spark.implicits._

    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://hadoop101:3306")
      .option("dbtable", "hive_dwd.topn")
      .option("user", "root")
      .option("password", "wsx123$%^")
      .load()

        jdbcDF.show()

    jdbcDF.filter('domain === "www.google.com")
      .write.format("jdbc")
      .option("url", "jdbc:mysql://hadoop101:3306")
      .option("dbtable", "hive_dwd.topn_2")
      .option("user", "root")
      .option("password", "wsx123$%^")
      .save()
  }


  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    jdbc(spark)
    spark.stop()
  }
}

结果：写回MySQL

mysql> show tables;
+--------------------+
| Tables_in_hive_dwd |
+--------------------+
| stat               |
| topn               |
| topn_2             |
+--------------------+
3 rows in set (0.00 sec)

mysql> select * from topn_2;
+----------------+------+------+
| domain         | url  | cnt  |
+----------------+------+------+
| www.google.com | url2 |    2 |
| www.google.com | url6 |    7 |
| www.google.com | url1 |    1 |
| www.google.com | url8 |    7 |
+----------------+------+------+
4 rows in set (0.00 sec)

mysql>

但是按照上面写 是不是太恶心了 参数 全都写死的 
通过读取配置文件的方式  ：有很多种写法  这里列出一个

object SparkSessionApp2 {

    def jdbc2(spark: SparkSession) = {
      import spark.implicits._

      val config = ConfigFactory.load()
      val url = config.getString("db.default.url")
      val user = config.getString("db.default.user")
      val password = config.getString("db.default.password")
      val srcTable = config.getString("db.default.srctable")
      val targetTable = config.getString("db.default.targettable")

      val jdbcDF = spark.read
        .format("jdbc")
        .option("url", url)
        .option("dbtable", srcTable)
        .option("user", user)
        .option("password", password)
        .load()

          jdbcDF.show()

      jdbcDF.filter('domain === "www.google.com")
        .write.format("jdbc")
        .option("url", url)
        .option("dbtable", targetTable)
        .option("user", user)
        .option("password", password)
        .save()
    }

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkSessionApp")
      .getOrCreate()

    jdbc2(spark)
    spark.stop()
  }
}

在这里插入图片描述

mysql> select * from topn_3;
+----------------+------+------+
| domain         | url  | cnt  |
+----------------+------+------+
| www.google.com | url2 |    2 |
| www.google.com | url6 |    7 |
| www.google.com | url1 |    1 |
| www.google.com | url8 |    7 |
+----------------+------+------+
4 rows in set (0.00 sec)

mysql>