Spark应用程序Spark应用程序由一个驱动器进程和一组执行器进程组成。驱动进程运行main()函数位于集群中的一个节点上它负责三件事维护Spark应用程序的相关信息回应用户的程序或输入分析任务并分发给若干执行器进行处理。驱动器是必须的它是Spark应用程序的核心它在应用程序执行的整个生命周期中维护着所有相关信息。执行器负责执行驱动器分配给它的实际计算工作这意味着每个执行器只负责两件事执行由驱动器分配给它的代码并将该执行器的计算状态报告给运行驱动器的节点。Spark API的多语言支持ScalaSpark主要用Scala编写Java可以用Java编写Spark代码PythonPython几乎支持所有Scala支持的结构SQLSpark支持ANSI SQL2003标准的一个子集RSpark有两个常用的R库一个作为Spark核心的一部分SparkR另一个是R语言开源社区维护的包sparklyr启动Spark当以交互方式启动Spark时相当于隐式地创建了一个SparkSession来管理Spark应用程序当不是通过交互模式而是通过独立应用程序启动Spark时必须在应用程序代码中显式地创建SparkSession对象。DataFrameDataFrame是最常见的结构化API简单来说它是包含行和列的数据表说明这些列和列类型的一些规则被称为模式schema可以将DataFrame想象为具有多个命名列的电子表格但是我们熟悉的电子表格和Spark DataFrame是不同的电子表格位于一台计算机上而Spark DataFrame可以跨越数千台计算机。Spark有几个核心抽象DatasetDataFrameSQL表和RDD这些不同的抽象都表示分布式数据集合其中最简单和最有效的是DataFrame它支持所有语言数据分区为了让多个执行器并行地工作Spark将数据分解成多个数据块每个数据块叫做一个分区。分区是位于集群中的一台物理机上的多行数据的集合DataFrame的分区也说明了在执行过程中数据在集群中的物理分布。如果只有一个分区即使拥有数千个执行器Spark也只有一个执行器在处理数据。类似地如果有多个分区但只有一个执行器那么Spark仍然只有一个执行器在处理数据就是因为只有一个计算资源单位。转换操作转换操作是使用Spark表达业务逻辑的核心有两类转换操作第一类是指定窄依赖关系的转换操作第二类是指定宽依赖关系的转换操作具有窄依赖关系的转换操作我们称之为窄转换是每个输入分区仅决定一个输出分区的转换具有宽依赖关系的转换是每个输入分区决定了多个输出分区这种宽依赖关系的转换经常被称为洗牌shuffle操作它会在整个集群中执行互相交换分区数据的功能。如果是窄转换Spark将自动执行流水线处理这意味着如果我们在DataFrame上指定了多个过滤操作它们将全部在内存中执行而属于宽转换的shuffle操作不是这样当我们执行shuffle操作时Spark将结果写入磁盘。惰性评估惰性评估的意思就是等到绝对需要时才执行计算。在Spark中当用户表达一些对数据的操作时不是立即修改数据而是建立一个作用到原始数据的转换计划。Spark首先会将这个计划编译为可以在集群中高效运行的流水线式的物理执行计划然后等待直到最后时刻才开始执行代码。这会带来很多好处因为Spark可以优化整个从输入端到输出端的数据流。一个很好的例子就是DataFrame的谓词下推假设我们构建一个含有多个转换操作的Spark作业并在最后指定了一个过滤操作假设这个过滤操作只需要数据源中的某一行数据则最有效的方法是在最开始仅访问我们需要的单个记录Spark会通过自动下推这个过滤操作来优化整个物理执行计划。动作操作转换操作使我们能够建立逻辑转换计划为了触发计算我们需要运行一个动作操作action一个动作指示Spark在一系列转换操作后计算一个结果最简单的动作操作是count它计算一个DataFrame中的记录总数count不是唯一的动作有三类动作在控制台中查看数据的动作在某个语言中将数据汇集为原生对象的动作写入输出数据源的动作完整示例Spark可以从大量数据源中读取数据或写入数据为了读取这些数据需要用到与我们创建的SparkSession所关联的DataFrameReader还需要指定文件格式及设置其他选项。为了获取模式信息Spark会从文件中读取一些数据然后根据Spark支持的类型尝试解析读取这些行中的数据类型。当然也可以在读取数据时选择严格指定模式val flightData2015spark.read.option(inferSchema,true).option(header,true).csv(file:///data/2015-summary.csv)csv文件被读到一个dataframe里后又被转换为一个本地数组或行列表在默认情况下shuffle操作会输出200个shuffle分区我们将此值设置为5以减少shuffle输出分区的数量spark.conf.set(spark.sql.shuffle.partitions,5)我们构建的转换逻辑计划定义了DataFrame的血统这样在任何给定的时间点Spark都知道如何通过对输入数据执行之前的操作来重新计算任何分区这就是Spark编程模型的核心——函数式编程当数据转换保持不变时相同的输入始终导致相同的输出DataFrame和SQL不管使用什么语言Spark以完全相同的方式执行转换操作可以使用SQL或者DataFrame表达业务逻辑并且在实际运行代码之前Spark会将该逻辑编译到底层执行计划可以在解释计划中看到使用Spark SQL可以将任何DataFrame注册为数据表或视图临时表)并使用纯SQL对它进行查询。编写SQL查询或编写DataFrame代码并不会造成性能差异它们都会被”编译“成相同的底层执行计划。可以使用一个简单的方法将任何DataFrame放入数据表或视图中flightData2015.createOrReplaceTempView(flight_data_2015)请注意这些计划编译后是完全相同的基本执行计划再来执行一些更复杂的操作在数据中找到前五个目标国家spark.sql(select dest_country_name,sum(count) sc from flight_data_2015 group by dest_country_name order by sc desc limit 5).showflightData2015.groupBy(dest_country_name).sum(count).withColumnRenamed(sum(count),destination_total).sort(col(destination_total).desc).limit(5).show