据 IDC 预测，数据中心以及边缘计算和物联网 (IOT) 所产生的数据在未来七年内将增长为原来的五倍，达到 175 ZB。伴随数据的巨量增长，由 Apache 软件基金会提供的 Apache Spark 已成为分布式横向扩展数据处理的热门框架之一，可以在本地和云端数以百万计的服务器上运行。本章旨在介绍 Spark 框架，并说明其如何执行应用程序。
Apache Spark 是应用于大型数据处理的快速通用分析引擎，可在 Hadoop、Apache Mesos、Kubernetes 上运行，也可独立或在云端运行。Spark 提供高级运算符，让开发者能通过交互式 shell、笔记本或应用程序包使用 Scala、Python、R 或 SQL 语言轻松构建并行应用程序。
Spark 核心数据处理引擎之上存在多个用于 SQL 和 DataFrame、机器学习、GraphX、图形计算和流处理的库。用户可在来自各种数据源（例如 HDFS、Alluxio、Apache Cassandra、Apache HBase 或 Apache Hive）的海量数据集上结合使用这些库。
Spark 应用程序在集群节点上的执行程序进程内部作为并行任务运行，并在驱动程序中的 SparkSession 对象与集群上的资源或集群管理器（独立管理器、Mesos、YARN 或 Kubernetes）之间协调完成执行作业。
Spark 还可在单台计算机上运行，此称为本地模式。在本地模式下，驱动程序和任务在同一 Java 虚拟机内的线程中运行。 本地模式有利于执行原型设计、开发、调试和测试， 但此模式并不适于运行生产应用程序。
Spark DataFrame 是 org.apache.spark.sql.Row 对象的分布式数据集，跨集群内的多个节点进行分区，并具备并行操作性。DataFrame 表示具有行和列的数据表，类似于由 R 或 Python 语言编写的 DataFrame，但具备 Spark 优化功能。DataFrame 由多个分区组成，每个分区都是数据节点缓存中的一个行范围。
DataFrame 可以从不同数据源（如 csv、parquet、JSON 文件、Hive 表或外部数据库）进行构建。用户可以使用关系转换和 Spark SQL 查询对 DataFrame 进行操作。
Spark shell 或 Spark 笔记本则提供了一种交互式使用 Spark 的简单方式。您可以使用以下命令在本地模式下启动 shell：
$ /[installation path]/bin/spark-shell --master local[2]
随后，您可以将本章其余部分的代码输入 shell，以交互方式查看结果。在代码示例中，shell 的输出以结果开头。
为在应用程序驱动和集群管理器之间协调执行作业，请在程序中创建 SparkSession 对象，具体如以下代码示例所示：
val spark = SparkSession.builder.appName("Simple Application").master("local[2]").getOrCreate()
Spark 应用程序启动后，其将通过主 URL 连接到集群管理器。在创建 SparkSession 对象或提交 Spark 应用程序时，可以将主 URL 设置为集群管理器或 local[N]，以使用 N 个线程在本地运行。使用 Spark shell 或 Spark 笔记本时，SparkSession 对象已预先创建完毕，并可用作变量 spark。成功连接后，集群管理器将按照为集群内节点进行配置的方式来分配资源并启动执行程序进程。Spark 应用程序执行后，SparkSession 会将任务发送给执行程序以开始运行。
借助 SparkSession 读取方法，您可将文件中的数据读取到 DataFrame 中，以指定模式的文件类型、文件路径和输入选项。
import org.apache.spark.sql.types._
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
val schema =
StructType(Array(
StructField("vendor_id", DoubleType),
StructField("passenger_count", DoubleType),
StructField("trip_distance", DoubleType),
StructField("pickup_longitude", DoubleType),
StructField("pickup_latitude", DoubleType),
StructField("rate_code", DoubleType),
StructField("store_and_fwd", DoubleType),
StructField("dropoff_longitude", DoubleType),
StructField("dropoff_latitude", DoubleType),
StructField("fare_amount", DoubleType),
StructField("hour", DoubleType),
StructField("year", IntegerType),
StructField("month", IntegerType),
StructField("day", DoubleType),
StructField("day_of_week", DoubleType),
StructField("is_weekend", DoubleType)
))
val file = "/data/taxi_small.csv"
val df = spark.read.option("inferSchema", "false")
.option("header", true).schema(schema).csv(file)
result:
df: org.apache.spark.sql.DataFrame = [vendor_id: double, passenger_count:
double ... 14 more fields]
take 方法返回一个数组，其中包含此 DataFrame 中的对象，在本文示例中即为 org.apache.spark.sql.Row 类型的数组。
df.take(1)
result:
Array[org.apache.spark.sql.Row] =
Array([4.52563162E8,5.0,2.72,-73.948132,40.829826999999995,-6.77418915E8,-1.0,-73.969648,40.797472000000006,11.5,10.0,2012,11,13.0,6.0,1.0])
DataFrame 提供用于结构化数据处理的域特定语言 API，称作 transformation。transformation 操作将根据当前的 DataFrame 创建经转换的全新 DataFrame，而且采用延迟评估形式。当受到 action 的触发时，系统将执行 transformation，从而将结果返回至驱动或写入磁盘。在 action 完成运行并返回值后，DataFrame 将不再存储于内存中，除非将其缓存。通过调用 dataFrame.cache()，Spark 可以使用内存中的列式格式缓存 DataFrame。
下方列表列出一些常用的 DataFrame transformation 操作。
此 groupBy transformation 示例按一天中的每一小时对出租车 DataFrame 进行分组，然后通过 count 操作计算每小时的出租车行程数总和。show 操作则以表格格式打印出所得的 DataFrame 行。
df.groupBy("hour").count().show(4)
result:
+----+-----+
|hour|count|
+----+-----+
| 0.0| 12|
| 1.0| 49|
| 2.0| 658|
| 3.0| 742|
+----+-----+
下方列表列出一些常用的 DataFrame action 操作。
There are two types of DataFrame transformations, those with narrow dependencies and those with wide dependencies. Transformations with narrow dependencies do not have to move data between partitions when creating a new DataFrame from an existing one. An example narrow transformation is filter() which is used to filter the rows from a DataFrame based on the given SQL expression. The following example filters for the hour value = 0。
通过使用称为流水线 (pipelining) 的流程，内存中的 DataFrame 上可以执行多个窄依赖 transformation 操作，从而让此类操作变得非常高效。下方示例使用 filter 和 select 等窄依赖 transformation 操作来检索一天中 0 小时的出租车 fare_amounts。
// select 和 filter 是窄依赖 transformation
df.select($"hour", $"fare_amount").filter($"hour" === "0.0" ).show(2)
result:
+----+-----------+
|hour|fare_amount|
+----+-----------+
| 0.0| 10.5|
| 0.0| 12.5|
+----+-----------+
根据现有 DataFrame 创建新 DataFrame 时，具有宽依赖关系的 transformation 操作必须在分区之间移动数据，此过程称为 shuffle。shuffle 通过网络将数据发送到其他节点并写入磁盘，由此引起网络和磁盘之间的 I/O。宽依赖 transformation 操作的示例有：groupBy、agg、sortBy 和 orderBy。宽依赖 transformation 操作按小时值显示分组。
以下是按小时值进行分组并按小时对出租车行程进行计数的宽依赖 transformation 操作。
df.groupBy("hour").count().show(4)
result:
+----+-----+
|hour|count|
+----+-----+
| 0.0| 12|
| 1.0| 49|
| 2.0| 658|
| 3.0| 742|
+----+-----+
