在今天发表的另一篇博文中，我们展示了选择S3而不是HDFS的五大理由。随着简单有效的云存储系统（如amazons3）的主导地位，apachehadoop等内部部署系统的假设变得越来越清晰，有时甚至令人痛苦。apachespark用户需要对云存储系统进行快速且事务正确的写入。这篇博客文章将介绍Databricks如何通过比较当前Hadoop提交协议与Databricks的I/O（DBIO）事务提交协议的性能和正确性缺陷来实现这两个目标。Spark为什么需要事务写入？像apachespark这样的大规模数据处理框架通过将执行作业所需的工作分解为可检索的任务来实现容错。由于任务可能偶尔失败，Spark必须确保只有成功任务和作业的输出可见。从形式上讲，这是通过使用提交协议来实现的，该协议指定了在作业结束时应如何编写结果。Spark作业的作业提交阶段确保只有成功作业的输出对读者可见。根据我们的经验，当Spark在云本地设置中使用时，作业提交是性能和正确性问题的主要来源，例如，直接向S3这样的存储服务写入数据。为了更好地理解为什么需要作业提交，让我们比较一下如果Spark不使用提交协议，两种不同的失败场景。如果任务失败，它可能会将部分写入的数据保留在S3（或其他存储器）上。Spark调度程序将重新尝试该任务，这可能会导致重复的输出数据。如果整个作业失败，它可能会在S3上留下单个任务的部分结果。这两种情况中的任何一种都可能对企业极为不利。为了避免这些数据损坏的情况，Spark依赖于Hadoop的commit协议类，这些类首先将任务输出转移到临时位置，只在任务或作业完成时将数据移动到其最终位置。正如我们将要展示的，这些Hadoop协议不是为云本地设置而设计的，它迫使用户在性能和正确性之间做出选择。比较现有提交协议在介绍Databricks I/O（DBIO）事务提交之前，让我们先评估一下现有的Hadoop提交协议。提交协议可以从两个方面进行评估：性能–协议提交文件的速度有多快？当然，你希望你的工作尽快运行。事务性-作业能否只完成部分或损坏的结果？理想情况下，作业输出应该在事务上可见（即全部或全部不可见）。如果作业失败，则读取器不应观察损坏或部分输出。性能试验Spark附带了两个默认的Hadoop提交算法：版本1，在作业结束时将分阶段任务输出文件移到它们的最终位置；版本2，在单个作业任务完成时移动文件。我们来比较一下他们的表现。我们使用Databricks Community Edition上的Spark 2.1集群进行以下测试运行：//使用指定的Hadoop FileOutputCommitter版本附加10m行火花射程（10e6.托龙）.重新分区（100）。写入模式（"附加"）.选项（"mapreduce.fileoutputcommitter.algorithm.version"，"）.parquet（"/tmp/test-"）//使用v1与v2比较作业的总运行时间显示（Seq（（"Hadoop Commit V1"，（"Hadoop Commit V2"，））.toDF（"算法"，"时间"）因为它在任务完成后就开始并行移动文件，所以v2hadoop提交协议几乎比v1快5倍。这就是为什么在最新的Hadoop版本中，v2是默认的提交协议。交易性检验现在让我们来看看事务性。我们通过模拟由持续失败的任务导致的作业失败来评估这一点，这种情况在实践中经常发生，例如，如果特定文件中存在无法解析的不良记录，则可能会发生这种情况。可按如下方式进行：//将更多行追加到上一个输出目录火花射程（10000）.重新分区（7）.map{i=>如果（i==9999）{线程睡眠（5000）；抛出新的RuntimeException（"oops！"）}否则我}.写入.选项("mapreduce.fileoutputcommitter.algorithm.version"，"）.mode（"append"）.parquet（"/tmp/test-"）//比较失败作业中新添加的行数值newRowsV1=spark.read.拼花地板（"/tmp/test-1"）.count（）-10000000值newRowsV2=spark.read.拼花地板（"/tmp/test-2"）.count（）-10000000显示（Seq（（"Hadoop Commit V1"，newRowsV1），（"Hadoop Commit V2"，newRowsV2））.toDF（"algorithm"，"corrupted rows"））我们从经验上看到，虽然v2更快，但它也会在作业失败上留下部分结果，打破了事务性需求。实际上，这意味着对于链接的ETL作业，作业失败（即使重试成功）可能会复制下游作业的一些输入数据。这需要在使用链接的ETL作业时进行仔细的管理。DBIO事务提交没有任何妥协注意：从Databricks（spark2.1-db4）开始就可以使用这个特性所有Hadoop用户在向云存储写入数据时，都面临着性能可靠性的权衡，无论他们是否使用Spark。虽然v1更具事务性，但速度非常慢，因为在S3中移动文件的成本很高。然而，这种权衡并不是根本的，因此在Databricks中，我们为DBIO构建了一个新的事务提交协议，它与Databricks中的高可用性服务相协调。基本思路如下：当用户在作业中写入文件时，DBIO将为您执行以下操作。使用唯一事务id写入的标记文件。将文件直接写入其最终位置。当作业提交时，将事务标记为已提交。当用户读取文件时，DBIO将为您执行以下操作。检查它是否具有事务id和状态，如果事务尚未完成，则忽略文件，或者读入数据。这个简单的想法在不牺牲可靠性的前提下极大地提高了性能。我们从上面运行相同的性能测试，并与默认的Hadoop提交算法进行比较：在这个性能测试中，运行在Databricks上的Spark将击败两个Hadoop版本的提交协议。事实上，这种比较适用于所有类型的ETL工作负载。我们还对每个协议进行了理论正确性分析：协议是否保证在出现不同类型的故障时输出正确？无提交协议Hadoop提交V1Hadoop提交V2DBIO事务提交任务失败不是的是的是的作业失败（例如持续的任务失败）不是的不是的提交期间驱动程序失败不不不是的如表所示，Databricks的新事务提交协议在面对不同类型的失败时提供了强有力的保证。此外，通过强制执行正确性，这会给Spark用户带来一些额外的好处。安全任务推测-任务推测允许Spark在观察到某些任务执行异常缓慢时以推测方式启动任务。在当前的Hadoop提交协议中，由于文件冲突的可能性，在向S3写入时启用Spark任务推测是不安全的。使用事务提交，您可以使用火花。猜测在集群Spark配置中为true。投机可以减少零散任务对工作完成的影响，在某些情况下大大提高了绩效。原子文件覆盖-原子性覆盖一组现有文件有时很有用。今天，Spark通过首先删除数据集，然后执行产生新数据的作业来实现覆盖。这会中断所有当前的读卡器，并且不是容错的。使用事务提交，可以通过在提交时将文件标记为已删除，以原子方式"逻辑删除"文件。原子覆盖可以通过设置"spark.databricks.io.directoryCommit.enableLogicalDelete正确|错误"。这改善了那些同时访问相同数据集的用户体验。增强的一致性—我们的事务提交协议与其他数据块服务结合，有助于减轻S3最终可能因链接ETL作业而产生的一致性问题。在Databricks中启用DBIO事务提交从Spark 2.1-db4集群映像开始，可以使用以下SQL配置启用DBIO提交协议（文档）：%sql集合spark.sql.sources.commitProtocolClass类=com.databricks.io.委员会协议也可以在创建集群时通过设置相同的集群配置进行设置。我们还在Databricks Runtime 3.0beta中默认启用了DBIO事务提交，为所有Databricks客户带来了快速、正确的ETL。您可以在我们的文档中阅读有关此功能的更多信息。结论回顾一下，我们展示了现有的Hadoop提交协议在云本地设置中使用时强制进行性能完整性权衡。相比之下，DBIO事务提交提供了最好的性能和强大的正确性保证。在数据块上试试吧。免费试用Databricks。今天就开始吧