Spark SQL: Relational Data Processing in Spark

ABSTRACT

       Spark SQL是Apache Spark中的一个新模块，它将相关处理与Spark s函数式编程API集成在一起。基于我们对Shark的经验，Spark SQL允许Spark程序员利用关系处理(例如，声明查询和优化存储)的优点，并允许SQL用户调用Spark中的complexanalytics库(例如，机器学习)。与以前的系统相比，Spark SQL增加了两个主要特性：首先，它通过一个与过程性Spark代码集成的声明性DataFrame API，在关系处理和过程性处理之间提供了更紧密的集成。其次，它包括一个高度可扩展的优化器Catalyst，它是使用Scala编程语言的特性构建的，这使得添加可组合规则、控制共变性和定义扩展点变得很容易。使用Catalyst，我们构建了各种各样的特性(例如，JSON的模式推断、machine - learning类型和对外部数据库的查询联合)，以满足现代数据分析的复杂需求。我们看到Spark SQL是对SQL-on-Spark和Spark本身的进化。

Introduction

      大数据应用需要处理技术、数据源和存储格式的混合。最早为这些工作应用程式(如MapReduce)设计的系统为用户提供了一个强大的、但底层的过程式编程接口。编程这样的系统是繁重的，用户需要手动优化来实现高性能。因此，多个新系统试图通过向大数据提供关系接口来提供更高效的用户体验。诸如Pig、Hive、Dremel和shark[29,36,25,38]这样的系统都利用声明性查询来提供更丰富的自动优化。

      尽管关系系统的流行表明用户通常更喜欢编写声明性查询，关系方法不适用于许多大数据应用。首先，用户希望能够访问和访问各种数据源，这些数据源可能是半结构化或非结构化的，需要定制代码。其次，用户希望执行高级分析，如机器学习和图形处理，这在关系系统中很难表达。实际上，我们已经注意到，大多数数据管道都应该结合关系查询和复杂过程算法进行组合。不幸的是，到目前为止，这两类系统-关系和过程仍然很大程度上互不关联，迫使用户只能选择一种范式或另一种范式。

       本文描述了我们在SparkSQL中结合这两个模型所做的工作，Apache Spark中的一个重要新组件。Spark SQL建立在我们之前的SQL-on-Spark工作上，被叫做Shark。不再是强迫用户在关系API和过程API之间进行选择，而是允许用户无缝地混合这两者。

        Spark SQL通过两个贡献弥合了两个模型之间的差距。首先，Spark SQL提供了可以在外部数据源和Spark 的内置的分布式集合上执行关系操作的aDataFrame api。这个API类似于R中广泛使用的数据帧概念，但是这些数据操作是lazy的，因为可以实现关系之间的优化处理。其次，是支持大数据中广泛的数据源和算法。Spark SQL引入了一种新的可扩展优化器，称为catalyst。Catalyst可以方便地为机器学习等领域添加数据源、优化规则和数据类型。

        DataFrame API在Spark程序中提供了丰富的关系/过程集成。DataFrames是结构化记录的集合，可以使用Spark的过程式API对其进行操作，或者使用新的关系api来实现更丰富的优化。可以直接从Spark的Java / Python对象的内置分布式集合中创建它们，在现有的spark程序中启用关系处理。其他Spark组件（例如机器学习库）也采用并产生DataFrames。在许多常见的情况下，DataFrames比Spark s procedure API更方便、更高效。例如，它们使使用SQL语句轻松地一次计算多个聚合成为可能，在传统函数api中难以表达的东西。它们还以列格式自动存储数据，该格式比Java / Python对象紧凑得多。 最后，与R和Python中现有的数据框架API不同，SparkSQL中的DataFrame操作通过关系优化器Catalyst。

        为了实现Spark sql对多种数据源和分析工作的支持，我们设计了一个名为catalyst的可扩展查询优化器。Catalyst使用Scala编程语言的功能，例如模式匹配，以图灵完备的语言表达可组合的规则。它提供了一个转换树的通用框架，我们用它来执行分析和规划,和运行时代码生成。通过这个框架，Catalyst还可以扩展新的数据源，包括JSON等半结构化数据以及可以推送过滤器(如HBase)的智能数据存储;和用户自定义函数，以及机器学习等领域的用户定义类型。众所周知，函数式语言非常适合于构建编译器[37]，因此，它们使构建可扩展优化器变得很容易，这也许并不奇怪。我们确实发现Catalyst能够有效地帮助我们快速添加Spark SQL的功能，而且自从它发布以来，我们已经看到外部贡献者也可以很容易地添加它们。

        Spark SQL于2014年5月发布，现已成为Spark中最活跃的组件之一。 在撰写本文时，Apache Spark是最活跃的大数据处理开源项目，在过去的一年中有400多个贡献者。 Spark SQL已经部署在非常大规模的环境中。 例如，一家大型的互联网公司使用Spark SQL构建数据管道，并在具有100PB以上数据的8000个节点的群集上运行查询。 每个单独的查询通常在数十TB上运行。 此外，许多用户不仅将Spark SQL用于SQL查询，而且在将其与过程处理相结合的程序中采用Spark SQL。例如，运行Spark的托管服务Databricks Cloud的2/3客户在其他编程语言中使用Spark SQL。在性能方面，我们发现对于关系查询，Spark SQL与Hadoop上仅SQL的系统相比具有竞争优势。它的速度和内存效率也比用SQL表达的简单的Spark代码计算要快10倍。

        更一般地说，我们将Spark SQL看作是核心Spark API的重要演化。虽然Spark最初的函数式编程api相当通用，但它只提供了有限的自动优化机会。同时，Spark SQL使Spark可供更多用户访问，并改进了现有用户的优化。在Spark中，社区现在将Spark SQL 合并到更多api中。DataFrames是新的用于机器学习的“ ML管道” API中的标准数据表示形式，我们希望将其扩展到其他组件，例如GraphX和流技术。

本文首先介绍了Spark和Spark SQL的目标，然后，我们描述DataFrame API，the Catalyst优化器，以及我们在catalyst上构建的高级特性。

Background and Goals

Spark Overview

      Apache Spark是一个通用的集群计算引擎，具有Scala、Java和Python中的api，以及用于流、图形处理和机器学习的库。发布于2010年，据我们所知，它是使用最广泛的系统之一，具有与DryadLINQ[20]类似的语言集成API，是大数据处理最活跃的开源项目。Spark在2014年有超过400个贡献者，并由多个供应商打包。

       Spark提供了类似于其他recent systems[20,11]的函数式编程API，其中用户操作称为弹性分布式数据集(RDDs)[39]的分布式集合。每个RDD都是跨集群分区的Java或Python对象的集合，可以通过map、filter和reduce等操作操作RDDs，这些操作使用编程语言中的函数并将它们发送到集群中的节点。例如，下面的Scala代码计算文本文件中以ERROR开头的行数：

lines = spark.textFile("hdfs://...")

errors = lines.filter(s => s.contains("ERROR"))

println(errors.count())

        这段代码通过读取一个hdfs文件来创建一个名为lines的字符串的RDD，然后使用filterts将其转换为另一个RDD，即errors。然后，它对这些数据执行运算。

        RDDs是容错的，因词系统可以使用RDDs的沿袭图恢复丢失的数据(通过重新运行诸如上述filter之类的操作来重建丢失的分区)。它们也可以显式地缓存在内存或磁盘中，以支持迭代。

        关于API的最后一点注意事项是，RDDs是可延迟计算的。每个RDD代表一个计算数据集的逻辑计划，但是Spark会等到某些输出操作(比如count)启动计算之后。这允许引擎执行一些简单的查询优化，比如流水线操作。例如，在上面的例子中，Spark将通过应用过滤器和计算运行计数来从HDFS文件中读取行，因此他不需要具体化中间产物lines and errors结果。而这样的优化是非常有用的，它也是有限的，因为引擎不理解RDDs中的数据结构(which is arbitrary Java/Python objects)或者用户函数语义(包含任意代码)。

        我们在Spark上构建关系接口的第一次尝试是Shark，改良Apache Hive系统以在Spark上运行，并实现了传统的RDBMS优化，例如，通过Spark引擎进行列处理。而Shark则表现出良好的性能和与Spark项目集成的良好机会，但是她面临是那个重大挑战。首先，Shark只能用于查询存储在Hive目录中的外部数据，因此不适用于Spark程序中的数据的关系查询。其次，从Spark程序调用Shark的唯一方法是将SQL字符串放在一起，这对于在模块化程序中使用很不方便且容易出错。最后，Hive优化器是为MapReduceand难以扩展而量身定做的，这使得它很难构建新的特性，例如用于机器学习的数据类型或对新数据源的支持

Goals for Spark SQL

         有了Shark的经验，我们希望扩展关系处理，使其涵盖Spark中的本地RDDs和更广泛的数据源。我们为Spark SQL设置了以下目标：

1、支持在Spark程序(本地RDDs上)和外部数据源上使用程序员友好的API进行关系处理。

2、使用已建立的DBMS技术提供高性能。

3、轻松支持新数据源，包括半结构化数据和可用于查询联合的外部数据库。

4、支持扩展与先进的分析算法，如图形处理和机器学习。

Programming Interface

Spark SQL作为一个库运行在Spark之上，如图1所示，它公开了SQL接口，可以通过JDBC / ODBC或命令行控制台进行访问，以及集成到Spark支持的编程语言中的dataframe API。我们首先介绍DataFrame API，该API使用户可以混合过程和关系代码。但是，也可以通过udf在SQL中公开高级函数，例如允许通过业务智能工具调用它们。我们在第3.7节中讨论。

DataFrame API

       Spark SQL 的API中的主要抽象是一个数据框架，它是具有相同模式的分布式行集合。DataFrame等同于关系数据库中的表，并且还可以通过与Spark（RDD）中的“本地”分布式集合类似的方式进行操作。与RDDs不同，DataFrames跟踪它们的模式并支持各种关系操作，从而实现更优化的执行。

       可以从系统目录中的表（基于外部数据源）或从nativeJava / Python对象的现有RDD（第3.5节）构造DataFrame。一旦构造完成，它们就可以使用各种关系运算符进行管理，例如wher and groupby，它使用领域特定语言(DSL)中的表达式，类似于R和Python中的数据帧[32,30]。每个dataframe也可以被看作是一个Row对象的RDD，允许用户调用过程性的Spark api，比如map。

       最后，与传统的数据帧api不同，Spark数据流是惰性的，因为每个DataFrame对象代表一个计算数据集的逻辑计划，但是直到用户调用一个特殊的输出操作(比如assave)才会执行。这支持对用于构建DataFrame的所有操作进行丰富的优化。

        为了说明这一点，下面的Scala代码定义了一个DataFrame from atable in Hive，根据它派生出另一个DataFrame，并输出一个结果：

       在这段代码中，users and young are DataFrames。snippet users(“age”)< 21是数据帧DSL中的一个表达式，它被捕获为一个抽象语法树，而不是像传统的Spark API那样表示一个标量函数。最后，每个DataFrame simply表示一个逻辑计划(即，读取年龄< 21岁的用户表和筛选器。当用户调用作为输出操作的calls count时，Spark SQL构建一个物理计划来计算最终结果。这可能包括一些优化，比如只扫描数据的年龄列(如果数据的存储格式是柱状的)，或者甚至在数据源中使用索引来计算匹配的行。

        接下来，我们将介绍DataFrame API的详细信息。

Data Model

        Spark SQL为表和数据流使用基于Hive[19]的嵌套数据模型。它支持所有主要的SQL数据类型，包括布尔、整数、双精度、小数、字符串、日期和时间戳，包括复杂的(即。，非原子)数据类型：结构，数组，映射和联合。 复杂的数据类型也可以嵌套在一起以创建更强大的类型。与许多传统的dbms不同，Spark SQL为查询语言和API中的复杂数据类型提供了一流的支持。此外，Spark SQL还支持用户自定义的类型，如4.4.2节所述。

        通过使用这个类型系统，我们能够从各种来源和格式中精确地建模数据，包括Hive、relational databases、JSON和Java/Scala/Python中的原生对象。

DataFrame Operations

       用户可以使用与R数据帧[32]和python panda[30]类似的特定于DSL语言(DSL)对数据流执行关系操作。DataFrames支持所有常见的关系操作符，包括投影(select)、筛选(where)、联接和聚合(groupBy)。这些操作符都以expressionobject(表达式对象)为例，在一个有限的DSL中，Spark捕捉表达式的结构。例如，下面的代码计算每个部门的正式员工数量。

       在这里，employeesis一个DataFrame和employees（“ deptId”）是表示deptId列的表达式。表达式对象有许多返回新表达式的运算符，包括通常的比较运算符(例如，===for equal test，>for greater than)和算术运算符(+，-，等等)。它们还支持聚合，列入count(“name”)。所有这些运算中的所有运算符都会构建表达式的抽象语法树（AST），然后将其传递给Catalyst进行优化。这不同于本机的Spark API，后者采用的函数包含任意Scala / Java / Python代码，然后这些代码对于运行时引擎是不透明的。

       除了关系DSL之外，DataFrames还可以在系统目录中注册一个临时表，并使用SQL进行查询。下面的代码显示了一个示例：

       SQL有时可以方便地简洁地计算多个聚合，并且还允许程序通过JDBC / ODBC公开数据集。 目录中注册的DataFrame仍然是未实现的视图，因此可以在SQL和原始DataFrame表达式中进行优化。 但是，正如我们在3.6节中讨论的，DataFrames也可以实现。

DataFrames versus Relational Query Languages

       从表面上看，DataFrames提供与SQL和Pig [29]之类的关系查询语言相同的操作，我们发现，由于它们与完整的编程语言集成在一起，因此它们可以大大简化用户的使用。 例如，用户可以将其代码分解为Scala，Java或Python函数，这些函数在它们之间传递数据框以建立逻辑计划，并且在运行输出操作时仍将受益于整个计划的优化。 同样，开发人员可以使用控制结构（如if语句和循环）来构造其工作。 一位用户说，DataFrame API“像SQL一样简洁明了，但我可以命名中间结果，”是指如何更轻松地构建计算和调试中间步骤。

       为了简化DataFrames中的编程，我们还创建了API analyze logical plans eagerly(即，以确定表达式中使用的列名是否存在于底层表中，以及它们的数据类型是否合适)，即使查询结果是延迟计算的。只要用户键入无效的代码行，Spark SQL就会报告错误，而不是等到执行为止。这比使用大型SQL语句更容易。

Querying Native Datasets

       现实世界中的管道通常从异构源提取数据，并从不同的程序库运行各种各样的算法。为了与过程性Spark代码互操作，Spark SQL 允许用户直接针对编程语言的对象的RDD构造DataFrame。Spark SQL可以使用反射自动推断这些对象的Schema。在Scala和Java中，类型信息是从语言的类型系统中提取的（从JavaBeans和Scala案例类中提取），在Python中，由于动态类型系统，Spark SQL对数据集进行采样以执行模式推断。

         例如，下面的Scala代码根据Userobjects的RDD定义了一个DataFrame。 Spark SQL自动检测列的名称（“ name”和“ age”）和数据类型（字符串和整数）。

       最后，Spark SQL创建一个指向RDD的逻辑数据扫描操作符。这被编译为访问本机对象字段的物理运算符。需要注意的是，这与传统的对象关系映射(ORM)有很大的不同。ORM通常会进行昂贵的转换，从而将整个对象转换为不同的格式。相比之下，Spark SQL就地访问原始对象，仅提取每个查询中使用的字段。

      查询本机数据集的能力使用户可以在现有Spark程序中运行优化的关系操作。此外，它还简化了RDDs与外部结构化数据的组合。例如，我们可以使用Hive中的表来join userDF:

In-Memory Caching

       Like Shark before it，Spark SQL可以使用列存储在内存中实现(通常称为缓存)热点数据。与Spark的本机缓存仅将数据存储为JVM对象相比，列式缓存可以将内存压缩减少一个数量级，因为它采用了列压缩方案，例如字典编码和游程长度编码。缓存对于交互式查询和机器学习中常见的迭代算法特别有用。 可以通过在DataFrame上调用cache（）来调用它。

User-Defined Functions

       用户定义函数(udf)一直是数据库系统的一个重要扩展点。例如，MySQL依赖于udf来提供对JSON数据的基本支持。一个更高级的示例是MADLib使用UDF为Postgres和其他数据库系统实现机器学习算法。但是，数据库系统通常要求在与主要查询接口不同的单独编程环境中定义UDF。Spark SQL s DataFrame API支持udf的内联定义，无需其他数据库系统中复杂的打包和注册过程。 事实证明，此功能对于采用API至关重要。在Spark SQL中，可以通过传递Scala，Java或Python函数来内联注册UDF，这些函数可能会在内部使用完整的Spark API。例如，给定机器学习模型的模型对象，我们可以将其预测函数注册为UDF：

        一旦注册，UDF也可以由商业智能工具通过JDBC / ODBC接口使用。 除了可以按标量值进行操作的UDF（如此处所示）外，还可以像MADLib [12]中那样通过使用其名称来定义对整个表进行操作的UDF，并在其中使用分布式Spark API，从而向SQL用户公开高级分析功能。最后，由于UDF定义和查询执行是使用相同的通用语言（例如，Scala或Python）表示的，因此用户可以使用标准工具调试或分析整个程序。

        上面的示例演示了许多管道中的常见用例，即同时使用关系运算符和高级分析方法的情况，这些方法在SQL中难以表达.DataFrame API使开发人员可以无缝地混合使用这些方法。

Catalyst Optimizer

        为了实现Spark SQL，我们设计了一个新的可扩展优化器Catalyst，它基于Scala中的函数式编程结构。Catalyst的可扩展设计有两个目的。首先，我们想轻松地向Spark SQL添加新的优化技术和功能，尤其是解决“大数据”（例如，半结构化数据和高级分析）所特有的各种问题。其次，我们希望使外部开发人员能够扩展优化器，例如，通过添加特定于数据源的规则，这些规则可以将过滤或聚合推入外部存储系统，或者支持新的数据类型。 Catalyst支持基于规则的优化和基于成本的优化。

         尽管过去已经提出了可扩展的优化器，但它们通常需要使用复杂的领域特定语言来指定规则，并需要“优化器编译器”将规则转换为可执行代码。他导致了巨大的学习曲线和维护负担。比之下，Catalyst使用Scala编程语言的标准功能，例如模式匹配[14]，使开发人员可以使用完整的编程语言，同时仍可以轻松指定规则。功能语言的设计部分是为了构建编译器，因此我们发现Scala非常适合此任务。据我们所知，Catalyst是第一个基于这种语言构建的生产质量查询优化器。

Catalyst的核心包含一个通用库，用于表示树并应用规则对其进行操作。在此框架之上，我们构建了特定于关系查询处理的库。（例如：表达式、逻辑查询计划)，以及一些处理查询执行的不同阶段的规则集:分析、逻辑优化、物理规划和将查询的各个部分编译为Java字节码的代码生成。对于后者，我们使用了另一个Scala功能，即准引用[34]，它使在运行时从可组合表达式生成代码变得容易。最后，Catalyst提供了几个公共扩展点，包括外部数据源和用户定义的类型。

Trees

       Catalyst中的主要数据类型是由节点对象组成的树。每个节点都有一个节点类型和零个或多个子节点。新的节点类型在Scala中定义为TreeNode类的子类，这些对象是不可变的，可以使用功能转换对其进行操作，如下一节所述。

       作为一个简单的示例，假设对于一种非常简单的表达语言，我们具有以下三个节点类：

       - Literal(value: Int): 一个常数值

       - Attribute(name: String):输入行的属性  

       - Add(left: TreeNode, right: TreeNode): 两个表达式的和

        These classes can be used to build up trees; for example, the treefor the expressionx+(1+2), shown in Figure 2, would be representedin Scala code as follows:

        Add(Attribute(x), Add(Literal(1), Literal(2)))

Rules

       可以使用规则来操纵树，从一棵树到另一棵树的功能。虽然规则可以在其输入树上运行任意代码（假设该树只是一个Scala对象），但最常见的方法是使用一组模式匹配函数来查找和替换具有特定结构的子树。模式匹配是许多函数式语言的一个特性，在Catalyst中，树提供了一种转换方法，该方法将模式匹配功能递归地应用于树的所有节点，并将与每个模式匹配的节点转换为结果。例如，我们可以实现一个规则，该规则将常量之间的Add操作折叠如下：

tree.transform {case Add(Literal(c1), Literal(c2)) => Literal(c1+c2)}

      将其应用于图2中的树forx +（1 + 2）将产生新树x + 3。这里的case关键字是Scala的标准模式匹配语法，可以用于匹配对象的类型，也可以为提取的值(这里是c1和c2)命名。传递给部分函数转换的模式匹配表达式，这意味着它只需要匹配所有可能的输入树的子集。Catalyst将测试给定规则适用于树的哪些部分，自动跳过并下降到不匹配的子树中。这种能力意味着规则只需要对应用给定优化的树进行推理，而不需要对不匹配的树进行推理。

      规则（通常与Scala模式匹配）可以在同一转换调用中匹配多个模式，使其非常简洁，一次即可实现多个转换：

       实际上，规则可能需要执行多次才能完全转换树。 Catalyst将规则编组成批，并执行每个批处理，直到到达固定点，即直到应用规则树停止更改后为止。将规则运行到固定点意味着每个规则可以简单且独立，并且最终仍然会对树产生更大的全局影响。 在上面的示例中，重复应用将恒定折叠较大的树，例如（x + 0）+（3 + 3）。 作为另一个示例，第一批可能会分析表达式以将类型分配给所有属性，而第二批可能会使用这些类型进行恒定折叠。 每次批处理之后，开发人员还可以对新树进行完整性检查（例如，查看是否为所有属性分配了类型），通常也可以通过递归匹配来编写。

       最后，规则条件及其主体可以包含任意Scala代码。 对于优化器而言，这使Catalyst比领域特定语言具有更大的功能，同时对于简单的规则也保持简洁。

      根据我们的经验，不可变树上的功能转换使整个优化器非常容易推理和调试，尽管我们尚未利用此功能，但它们还可以在优化器中实现并行化。

Using Catalyst in Spark SQL

       我们在四个阶段中使用Catalyst的通用树转换框架，如图3所示：（1）分析逻辑计划以解析引用，（2）逻辑计划优化，（3）物理计划，以及（4）代码生成以编译查询的各个部分 在物理计划阶段，Catalyst可能会生成多个计划并根据成本进行比较。 所有其他阶段都是纯粹基于规则的。 每个阶段使用不同类型的树节点。 Catalyst包括用于表达式，数据类型以及逻辑和物理运算符的节点库。 现在我们描述每个阶段。

Analysis

       Spark SQL以要计算的关系开头，该关系可以是SQL解析器返回的抽象语法树（AST），也可以是使用API​​构造的DataFrame对象。 在这两种情况下，关联都可能包含未解决的属性引用或关系：例如，在SQL查询中，SELECT col FROM sales中，col的类型，甚至它是否是有效的列名，直到我们查询表格时才知道。 如果我们不知道属性的类型或未将其与输入表（或别名）匹配，则该属性称为未解析的。 Spark SQL使用Catalyst规则和Catalog对象，该对象跟踪所有数据源中的表以解析这些属性。 首先，通过构建具有未绑定属性和数据类型的“未解决的逻辑计划”树，然后应用执行以下操作的规则：

1、从目录中按名称查找关系

2、将命名属性（例如col）映射到所提供的输入，以提供操作员的子级

3、确定哪些属性引用相同的值以赋予它们唯一的ID（以后可以优化表达式，例如col = col）

4、通过表达式传播和强迫类型：例如，我们不知道1 + coluntil的类型，我们已经解析了coland可能将其子表达式转换为兼容类型。

Logical Optimization

       逻辑优化阶段将基于规则的标准优化应用于逻辑计划。 这些包括恒定折叠，谓词下推，投影修剪，空传播，布尔表达式简化和其他规则。总的来说，我们发现为各种情况添加规则极其简单。 例如，当我们将固定精度DECIMAL类型添加到SparkSQL时，我们想以小精度优化诸如DECIMAL上的求和和平均值之类的聚合。 用了12行代码编写了一条规则，该规则在SUM和AVG表达式中找到这样的小数，然后将它们广播到未缩放的64位长，在其上进行汇总，然后将结果转换回去。 以下是仅优化SUM表达式的此规则的简化版本：

        另一个例子，作为另一个例子，一个12行的规则将简单的正则表达式优化为String.startsWithorString.containscalls。使用任意Scala代码规则的自由使得这些优化方式更易于表达，而这些优化方式超出了模式匹配子树的结构。 总的来说，逻辑优化规则是800行代码。