1. 实验室名称：

大数据实验教学系统

2. 实验项目名称：

Spark SQL的数据分析应用

3. 实验学时：

4. 实验原理：

DataFrame API 的设计目的是在数据集中操作或转换单个行，如过滤或分组。如果我们想要转换一个数据集中的每一行的列的值，例如将字符串从大写字母转换成驼峰命名形式，那么我们将使用一个函数来实现这一点。函数基本上就是应用于列的方法。Spark SQL 提供了一组通常需要的函数，同时也提供了创建新函数的简单方法。

尽管 Spark SQL 为大多数常见用例提供了大量的内置函数，但总会有一些情况下，这些功能都不能提供您的用例所需要的功能。Spark SQL 提供了一个相当简单的工具来编写用户定义的函数（UDF），并在 Spark 数据处理逻辑或应用程序中使用它们，就像使用内置函数一样。UDFs 实际上是您可以扩展 Spark 的功能以满足您的特定需求的一种方式。我最喜欢Spark 的另一件事是 UDFs 可以用 Python、Java 或 Scala 来写，它们可以利用和集成任何必要的库。因为您能够使用您最熟悉的编程语言来编写 UDFs，所以开发和测试 UDFs 是非常简单和快速的。

5. 实验目的：

掌握Spark SQL常用内置函数的使用。

掌握Spark SQL自定义函数(UDF)的使用。

6. 实验内容：

1、学习Spark SQL常用内置函数的使用。

2、学习Spark SQL自定义函数(UDF)的使用。

7. 实验器材（设备、虚拟机名称）：

硬件：x86_64 ubuntu 16.04服务器

软件：JDK 1.8，Spark-2.3.2，Hadoop-2.7.3，zeppelin-0.8.1

8. 实验步骤：

8.1 启动Spark与zeppelin

1、在终端窗口下，输入以下命令，分别启动Spark集群和Zeppelin服务器：

1.  $ cd /opt/spark
2.  $ ./sbin/start-all.sh
3.  $ zeppelin-daemon.sh start

2、启动浏览器，打开zeppelin notebook首页，点击【Create new note】链接，创建一个新的笔记本，如下图所示：

8.2 Spark SQL内置函数

Spark SQL提供了很多内置函数，包括处理日期时间函数、处理字符串函数和数学处理函数。

1、处理日期时间函数。日期和时间转换函数有助于将字符串转换为日期、时间戳或 Unix 时间戳，反之亦然。

在内部，它使用 Java 日期格式模式语法。这些函数使用的默认的日期格式是 yyyy-mm-dd HH:mm:ss。

下面的示例显示了将字符串类型的日期和时间戳转换为 Spark date 和 timestamp 类型。

在zeppelin中执行如下代码:

1.  //日期数据，最后两列不遵循默认日期格式
2.  val testDate = Seq((1, "2018-01-01", "2018-01-01 15:04:58:865", "01-01-2018", "12-05-2017 45:50"))
3.      　
4.  //添加列名
5.  val testDateTSDF = testDate.toDF("id","date", "timestamp","date_str", "ts_str")
6.      　
7.  // 将这些字符串转换为 date、timestamp 和 unix timestamp，并指定一个自定义的 date 和 timestamp格式
8.  val testDateResultDF= testDateTSDF.select(
9.      to_date('date).as("date1"), 
10.     to_timestamp('timestamp).as("ts1"), 
11.     to_date('date_str,"MM-dd-yyyy").as("date2"), 
12.     to_timestamp('ts_str,"MM-dd-yyyy mm:ss").as("ts2"), 
13.     unix_timestamp('timestamp).as("unix_ts"))
14.     　
15. //输出DataFrame中的数据结构信息，即为schema
16. testDateResultDF.printSchema
17.     　
18. //输出DataFrame
19. testDateResultDF.show

执行以上代码，输出结果如下:

1.   |-- date1: date (nullable = true)
2.   |-- ts1: timestamp (nullable = true)
3.   |-- date2: date (nullable = true)
4.   |-- ts2: timestamp (nullable = true)
5.   |-- unix_ts: long (nullable = true)
6.      　
7.  +----------+----+----------+-------------------+----------+
8.  |     date1| ts1|     date2|                ts2|   unix_ts|
9.  +----------+----+----------+-------------------+----------+
10. |2018-01-01|null|2018-01-01|2017-12-05 00:45:50|1514790298|
11. +----------+----+----------+-------------------+----------+

将日期或时间戳转换为时间字符串是很容易的，方法是使用 date_format 函数和定制日期格式，或者使用 from_unixtime 函数将 Unix 时间戳（以秒为单位）转换成字符串。

在zeppelin中执行如下代码：

1.  // 将日期、时间戳和 Unix 时间戳转换成字符串 
2.  testDateResultDF.select(date_format('date1,"dd-MM-YYYY").as("date_str"), 
3.                          date_format('ts1,"dd-MM-YYYY HH:mm:ss").as("ts_str"),
4.                          from_unixtime('unix_ts,"dd-MM-YYYY HH:mm:ss").as("unix_ts_str")).show

执行以上代码，输出结果如下:

1.  +----------+------+-------------------+
2.  |  date_str|ts_str|        unix_ts_str|
3.  +----------+------+-------------------+
4.  |01-01-2018|  null|01-01-2018 15:04:58|
5.  +----------+------+-------------------+

在处理时间序列数据（time-series data）时，能够提取日期或时间戳值的特定字段（如年、月、小时、分钟和秒）的能力是非常方便的。例如，当需要按季度、月或周对所有股票交易进行分组时，就可以从交易日期提取该信息，并按这些值分组。下面的代码展示了如何从日期或时间戳中提取字段。

在zeppelin中执行如下代码：

1.  // 从一个日期值中提取指定的字段 
2.  val valentimeDateDF=Seq(("2018-02-14 05:35:55")).toDF("date") 
3.      　
4.  valentimeDateDF.select(
5.      year('date).as("year"), 
6.      quarter('date).as("quarter"), 
7.      month('date).as("month"), 
8.      weekofyear('date).as("woy"), 
9.      dayofmonth('date).as("dom"), 
10.     dayofyear('date).as("doy"), 
11.     hour('date).as("hour"), 
12.     minute('date).as("minute"), 
13.     second('date).as("second")
14. ).show

结果：

1.  +----+-------+-----+---+---+---+----+------+------+
2.  |year|quarter|month|woy|dom|doy|hour|minute|second|
3.  +----+-------+-----+---+---+---+----+------+------+
4.  |2018|      1|    2|  7| 14| 45|   5|    35|    55|
5.  +----+-------+-----+---+---+---+----+------+------+

2、处理字符串函数。SparkSQL 内置的字符串函数提供了操作这类列的通用和强大的方法。

有很多方法可以转换字符串。最常见的是去空格、填充、大写、小写和连接。下面的代码展示了使用各种内置字符串函数转换字符串的各种方法。

在zeppelin中执行如下代码：

1.  //原始数据
2.  val sparkDF=Seq((" Spark ")).toDF("name")
3.  //原始数据展示
4.  sparkDF.show
5.      　
6.  //trimming 
7.  sparkDF.select(trim('name).as("trim"),ltrim('name).as("ltrim"),rtrim('name).as("rtrim")).show
8.      　
9.  // 用给定的 pad 字符串将字符串填充到指定长度 // 首先去掉"Spark"前后的空格，然后填充到 8 个字符长 
10. sparkDF.select(trim('name).as("trim"))
11.        .select(lpad('trim, 8, "-").as("lpad"),rpad('trim, 8, "=").as("rpad"))
12.        .show
13.     　
14. // 使用 concatenation,uppercase,lowercase 和 reverse 转换一个字符串 
15. val sparkAwesomeDF= Seq(("Spark","is","awesome")).toDF("subject","verb","adj") 
16. sparkAwesomeDF.select(concat_ws(" ",'subject,'verb, 'adj).as("sentence"))
17.               .select(lower('sentence).as("lower"), 
18.                       upper('sentence).as("upper"), 
19.                       initcap('sentence).as("initcap"), 
20.                       reverse('sentence).as("reverse")
21.                       )
22.               .show
23.     　
24. // 从一个字符转换到另一个字符 
25. sparkAwesomeDF.select('subject,translate('subject,"ar", "oc").as("translate")).show